enabling spotbugs for gcs-destinations submodule

airbytehq · Apr 3, 2024 · 47c584c · 47c584c
1 parent 23df75f
commit 47c584c
Show file tree

Hide file tree

Showing 12 changed files with 49 additions and 48 deletions.
diff --git a/.../kotlin/io/airbyte/cdk/integrations/standardtest/destination/DestinationAcceptanceTest.kt b/.../kotlin/io/airbyte/cdk/integrations/standardtest/destination/DestinationAcceptanceTest.kt
@@ -178,8 +178,8 @@ abstract class DestinationAcceptanceTest {
     @Throws(Exception::class)
     protected abstract fun retrieveRecords(
         testEnv: TestDestinationEnv?,
-        streamName: String?,
-        namespace: String?,
+        streamName: String,
+        namespace: String,
         streamSchema: JsonNode
     ): List<JsonNode>
 
@@ -1454,7 +1454,7 @@ abstract class DestinationAcceptanceTest {
             false
         )
         val destinationOutput =
-            retrieveRecords(testEnv, stream.name, getDefaultSchema(config), stream.jsonSchema)
+            retrieveRecords(testEnv, stream.name, getDefaultSchema(config)!!, stream.jsonSchema)
         // Remove state message
         secondSyncMessagesWithNewFields.removeIf {
             airbyteMessage: io.airbyte.protocol.models.v0.AirbyteMessage ->

diff --git a/airbyte-cdk/java/airbyte-cdk/gcs-destinations/build.gradle b/airbyte-cdk/java/airbyte-cdk/gcs-destinations/build.gradle
@@ -20,8 +20,6 @@ compileTestFixturesKotlin {
     }
 }
 
-spotbugsTestFixtures.enabled = false
-
 dependencies {
     implementation project(':airbyte-cdk:java:airbyte-cdk:airbyte-cdk-dependencies')
     implementation project(':airbyte-cdk:java:airbyte-cdk:airbyte-cdk-core')

diff --git a/...in/io/airbyte/cdk/integrations/destination/gcs/GcsAvroParquetDestinationAcceptanceTest.kt b/...in/io/airbyte/cdk/integrations/destination/gcs/GcsAvroParquetDestinationAcceptanceTest.kt
@@ -90,13 +90,14 @@ abstract class GcsAvroParquetDestinationAcceptanceTest(s3Format: S3Format) :
             else fieldDefinition["type"]
         val airbyteTypeProperty = fieldDefinition["airbyte_type"]
         val airbyteTypePropertyText = airbyteTypeProperty?.asText()
-        return Arrays.stream(JsonSchemaType.entries.toTypedArray())
+        return JsonSchemaType.entries
+            .toTypedArray()
             .filter { value: JsonSchemaType ->
                 value.jsonSchemaType == typeProperty.asText() &&
                     compareAirbyteTypes(airbyteTypePropertyText, value)
             }
-            .map(JsonSchemaType::avroType)
-            .collect(Collectors.toSet())
+            .map { it.avroType }
+            .toSet()
     }
 
     private fun compareAirbyteTypes(
@@ -126,8 +127,8 @@ abstract class GcsAvroParquetDestinationAcceptanceTest(s3Format: S3Format) :
 
     @Throws(Exception::class)
     protected abstract fun retrieveDataTypesFromPersistedFiles(
-        streamName: String?,
-        namespace: String?
+        streamName: String,
+        namespace: String
     ): Map<String?, Set<Schema.Type?>?>
 
     protected fun getTypes(record: GenericData.Record): Map<String, Set<Schema.Type>> {

diff --git a/...otlin/io/airbyte/cdk/integrations/destination/gcs/GcsBaseAvroDestinationAcceptanceTest.kt b/...otlin/io/airbyte/cdk/integrations/destination/gcs/GcsBaseAvroDestinationAcceptanceTest.kt
@@ -35,17 +35,17 @@ abstract class GcsBaseAvroDestinationAcceptanceTest :
     @Throws(Exception::class)
     override fun retrieveRecords(
         testEnv: TestDestinationEnv?,
-        streamName: String?,
-        namespace: String?,
+        streamName: String,
+        namespace: String,
         streamSchema: JsonNode
     ): List<JsonNode> {
-        val nameUpdater = getFieldNameUpdater(streamName!!, namespace, streamSchema)
+        val nameUpdater = getFieldNameUpdater(streamName, namespace, streamSchema)
 
-        val objectSummaries = getAllSyncedObjects(streamName, namespace)
+        val objectSummaries = getAllSyncedObjects(streamName, namespace!!)
         val jsonRecords: MutableList<JsonNode> = LinkedList()
 
-        for (objectSummary in objectSummaries!!) {
-            val `object` = s3Client!!.getObject(objectSummary!!.bucketName, objectSummary.key)
+        for (objectSummary in objectSummaries) {
+            val `object` = s3Client.getObject(objectSummary.bucketName, objectSummary.key)
             DataFileReader<GenericData.Record>(
                     SeekableByteArrayInput(`object`.objectContent.readAllBytes()),
                     GenericDatumReader<GenericData.Record>()
@@ -67,8 +67,8 @@ abstract class GcsBaseAvroDestinationAcceptanceTest :
 
     @Throws(Exception::class)
     override fun retrieveDataTypesFromPersistedFiles(
-        streamName: String?,
-        namespace: String?
+        streamName: String,
+        namespace: String
     ): Map<String?, Set<Schema.Type?>?> {
         val objectSummaries = getAllSyncedObjects(streamName, namespace)
         val resultDataTypes: MutableMap<String?, Set<Schema.Type?>?> = HashMap()

diff --git a/...kotlin/io/airbyte/cdk/integrations/destination/gcs/GcsBaseCsvDestinationAcceptanceTest.kt b/...kotlin/io/airbyte/cdk/integrations/destination/gcs/GcsBaseCsvDestinationAcceptanceTest.kt
@@ -40,8 +40,8 @@ abstract class GcsBaseCsvDestinationAcceptanceTest : GcsDestinationAcceptanceTes
     @Throws(IOException::class)
     override fun retrieveRecords(
         testEnv: TestDestinationEnv?,
-        streamName: String?,
-        namespace: String?,
+        streamName: String,
+        namespace: String,
         streamSchema: JsonNode
     ): List<JsonNode> {
         val objectSummaries = getAllSyncedObjects(streamName, namespace)

diff --git a/...tlin/io/airbyte/cdk/integrations/destination/gcs/GcsBaseJsonlDestinationAcceptanceTest.kt b/...tlin/io/airbyte/cdk/integrations/destination/gcs/GcsBaseJsonlDestinationAcceptanceTest.kt
@@ -36,8 +36,8 @@ abstract class GcsBaseJsonlDestinationAcceptanceTest :
     @Throws(IOException::class)
     override fun retrieveRecords(
         testEnv: TestDestinationEnv?,
-        streamName: String?,
-        namespace: String?,
+        streamName: String,
+        namespace: String,
         streamSchema: JsonNode
     ): List<JsonNode> {
         val objectSummaries = getAllSyncedObjects(streamName, namespace)

diff --git a/...in/io/airbyte/cdk/integrations/destination/gcs/GcsBaseParquetDestinationAcceptanceTest.kt b/...in/io/airbyte/cdk/integrations/destination/gcs/GcsBaseParquetDestinationAcceptanceTest.kt
@@ -37,17 +37,17 @@ abstract class GcsBaseParquetDestinationAcceptanceTest :
     @Throws(IOException::class, URISyntaxException::class)
     override fun retrieveRecords(
         testEnv: TestDestinationEnv?,
-        streamName: String?,
-        namespace: String?,
+        streamName: String,
+        namespace: String,
         streamSchema: JsonNode
     ): List<JsonNode> {
-        val nameUpdater = getFieldNameUpdater(streamName!!, namespace, streamSchema)
+        val nameUpdater = getFieldNameUpdater(streamName, namespace, streamSchema)
 
         val objectSummaries = getAllSyncedObjects(streamName, namespace)
         val jsonRecords: MutableList<JsonNode> = LinkedList()
 
-        for (objectSummary in objectSummaries!!) {
-            val `object` = s3Client!!.getObject(objectSummary!!.bucketName, objectSummary.key)
+        for (objectSummary in objectSummaries) {
+            val `object` = s3Client!!.getObject(objectSummary.bucketName, objectSummary.key)
             val uri = URI(String.format("s3a://%s/%s", `object`.bucketName, `object`.key))
             val path = Path(uri)
             val hadoopConfig = GcsParquetWriter.getHadoopConfig(config)
@@ -73,14 +73,14 @@ abstract class GcsBaseParquetDestinationAcceptanceTest :
 
     @Throws(Exception::class)
     override fun retrieveDataTypesFromPersistedFiles(
-        streamName: String?,
-        namespace: String?
+        streamName: String,
+        namespace: String
     ): Map<String?, Set<Schema.Type?>?> {
         val objectSummaries = getAllSyncedObjects(streamName, namespace)
         val resultDataTypes: MutableMap<String?, Set<Schema.Type?>?> = HashMap()
 
-        for (objectSummary in objectSummaries!!) {
-            val `object` = s3Client!!.getObject(objectSummary!!.bucketName, objectSummary.key)
+        for (objectSummary in objectSummaries) {
+            val `object` = s3Client!!.getObject(objectSummary.bucketName, objectSummary.key)
             val uri = URI(String.format("s3a://%s/%s", `object`.bucketName, `object`.key))
             val path = Path(uri)
             val hadoopConfig = getHadoopConfig(config)

diff --git a/...xtures/kotlin/io/airbyte/cdk/integrations/destination/gcs/GcsDestinationAcceptanceTest.kt b/...xtures/kotlin/io/airbyte/cdk/integrations/destination/gcs/GcsDestinationAcceptanceTest.kt
@@ -30,6 +30,7 @@ import org.joda.time.DateTime
 import org.joda.time.DateTimeZone
 import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.Test
+import org.mockito.Mockito.mock
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
@@ -46,9 +47,10 @@ import org.slf4j.LoggerFactory
 abstract class GcsDestinationAcceptanceTest(protected val outputFormat: S3Format) :
     DestinationAcceptanceTest() {
     protected var configJson: JsonNode? = null
-    protected lateinit var config: GcsDestinationConfig
-    protected lateinit var s3Client: AmazonS3
-    protected lateinit var nameTransformer: NamingConventionTransformer
+    // Not a big fan of those mocks(). Here to make spotbugs happy
+    protected var config: GcsDestinationConfig = mock()
+    protected var s3Client: AmazonS3 = mock()
+    protected var nameTransformer: NamingConventionTransformer = mock()
     protected var s3StorageOperations: S3StorageOperations? = null
 
     protected val baseConfigJson: JsonNode
@@ -96,23 +98,23 @@ abstract class GcsDestinationAcceptanceTest(protected val outputFormat: S3Format
 
     /** Helper method to retrieve all synced objects inside the configured bucket path. */
     protected fun getAllSyncedObjects(
-        streamName: String?,
-        namespace: String?
+        streamName: String,
+        namespace: String
     ): List<S3ObjectSummary> {
-        val namespaceStr = nameTransformer!!.getNamespace(namespace!!)
-        val streamNameStr = nameTransformer!!.getIdentifier(streamName!!)
+        val namespaceStr = nameTransformer.getNamespace(namespace)
+        val streamNameStr = nameTransformer.getIdentifier(streamName)
         val outputPrefix =
             s3StorageOperations!!.getBucketObjectPath(
                 namespaceStr,
                 streamNameStr,
                 DateTime.now(DateTimeZone.UTC),
-                config!!.pathFormat!!
+                config.pathFormat!!
             )
         // the child folder contains a non-deterministic epoch timestamp, so use the parent folder
         val parentFolder = outputPrefix.substring(0, outputPrefix.lastIndexOf("/") + 1)
         val objectSummaries =
             s3Client
-                .listObjects(config!!.bucketName, parentFolder)
+                .listObjects(config.bucketName, parentFolder)
                 .objectSummaries
                 .stream()
                 .filter { o: S3ObjectSummary -> o.key.contains("$streamNameStr/") }

diff --git a/.../kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseAvroDestinationAcceptanceTest.kt b/.../kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseAvroDestinationAcceptanceTest.kt
@@ -39,8 +39,8 @@ abstract class S3BaseAvroDestinationAcceptanceTest protected constructor() :
     @Throws(Exception::class)
     override fun retrieveRecords(
         testEnv: TestDestinationEnv?,
-        streamName: String?,
-        namespace: String?,
+        streamName: String,
+        namespace: String,
         streamSchema: JsonNode
     ): List<JsonNode> {
         val nameUpdater =

diff --git a/...s/kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseCsvDestinationAcceptanceTest.kt b/...s/kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseCsvDestinationAcceptanceTest.kt
@@ -36,8 +36,8 @@ abstract class S3BaseCsvDestinationAcceptanceTest : S3DestinationAcceptanceTest(
     @Throws(IOException::class)
     override fun retrieveRecords(
         testEnv: TestDestinationEnv?,
-        streamName: String?,
-        namespace: String?,
+        streamName: String,
+        namespace: String,
         streamSchema: JsonNode
     ): List<JsonNode> {
         val objectSummaries = getAllSyncedObjects(streamName, namespace)

diff --git a/...kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseJsonlDestinationAcceptanceTest.kt b/...kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseJsonlDestinationAcceptanceTest.kt
@@ -35,8 +35,8 @@ abstract class S3BaseJsonlDestinationAcceptanceTest protected constructor() :
     @Throws(IOException::class)
     override fun retrieveRecords(
         testEnv: TestDestinationEnv?,
-        streamName: String?,
-        namespace: String?,
+        streamName: String,
+        namespace: String,
         streamSchema: JsonNode
     ): List<JsonNode> {
         val objectSummaries = getAllSyncedObjects(streamName, namespace)

diff --git a/...tlin/io/airbyte/cdk/integrations/destination/s3/S3BaseParquetDestinationAcceptanceTest.kt b/...tlin/io/airbyte/cdk/integrations/destination/s3/S3BaseParquetDestinationAcceptanceTest.kt
@@ -29,8 +29,8 @@ abstract class S3BaseParquetDestinationAcceptanceTest protected constructor() :
     @Throws(IOException::class, URISyntaxException::class)
     override fun retrieveRecords(
         testEnv: TestDestinationEnv?,
-        streamName: String?,
-        namespace: String?,
+        streamName: String,
+        namespace: String,
         streamSchema: JsonNode
     ): List<JsonNode> {
         val nameUpdater =