enable spotbugs for CDK core submodule

airbyte-cdk/java/airbyte-cdk/core/build.gradle

@@ -50,9 +50,6 @@ compileKotlin {
     }
 }
 
-spotbugsTest.enabled = false
-spotbugsTestFixtures.enabled = false
-
 
 dependencies {
 

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/db/factory/CommonFactoryTest.kt

@@ -12,12 +12,13 @@ internal open class CommonFactoryTest {
     companion object {
         private const val DATABASE_NAME = "airbyte_test_database"
 
-        @JvmStatic protected lateinit var container: PostgreSQLContainer<*>
+        @JvmStatic
+        protected var container: PostgreSQLContainer<*> =
+            PostgreSQLContainer<Nothing>("postgres:13-alpine")
 
         @JvmStatic
         @BeforeAll
         fun dbSetup(): Unit {
-            container = PostgreSQLContainer<Nothing>("postgres:13-alpine")
             container.withDatabaseName(DATABASE_NAME).withUsername("docker").withPassword("docker")
             container!!.start()
         }

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/db/factory/DataSourceFactoryTest.kt

@@ -8,7 +8,6 @@ import io.airbyte.cdk.integrations.JdbcConnector
 import java.util.Map
 import javax.sql.DataSource
 import org.junit.jupiter.api.Assertions
-import org.junit.jupiter.api.BeforeAll
 import org.junit.jupiter.api.Test
 import org.mockito.Mockito
 import org.testcontainers.containers.MySQLContainer
@@ -182,7 +181,7 @@ internal class DataSourceFactoryTest : CommonFactoryTest() {
     @Test
     fun testCreatingADataSourceWithHostAndPort() {
         val dataSource =
-            DataSourceFactory.create(username, password, host, port!!, database, driverClassName)
+            DataSourceFactory.create(username, password, host, port, database, driverClassName)
         Assertions.assertNotNull(dataSource)
         Assertions.assertEquals(HikariDataSource::class.java, dataSource.javaClass)
         Assertions.assertEquals(
@@ -200,7 +199,7 @@ internal class DataSourceFactoryTest : CommonFactoryTest() {
                 username,
                 password,
                 host,
-                port!!,
+                port,
                 database,
                 driverClassName,
                 connectionProperties
@@ -218,14 +217,13 @@ internal class DataSourceFactoryTest : CommonFactoryTest() {
         val driverClassName = "Unknown"
 
         Assertions.assertThrows(RuntimeException::class.java) {
-            DataSourceFactory.create(username, password, host, port!!, database, driverClassName)
+            DataSourceFactory.create(username, password, host, port, database, driverClassName)
         }
     }
 
     @Test
     fun testCreatingAPostgresqlDataSource() {
-        val dataSource =
-            DataSourceFactory.createPostgres(username, password, host, port!!, database)
+        val dataSource = DataSourceFactory.createPostgres(username, password, host, port, database)
         Assertions.assertNotNull(dataSource)
         Assertions.assertEquals(HikariDataSource::class.java, dataSource.javaClass)
         Assertions.assertEquals(
@@ -249,24 +247,12 @@ internal class DataSourceFactoryTest : CommonFactoryTest() {
     companion object {
         private const val CONNECT_TIMEOUT = "connectTimeout"
 
-        var database: String? = null
-        lateinit var driverClassName: String
-        var host: String? = null
-        var jdbcUrl: String? = null
-        var password: String? = null
-        var port: Int? = null
-        var username: String? = null
-
-        @JvmStatic
-        @BeforeAll
-        fun setup(): Unit {
-            host = container!!.getHost()
-            port = container!!.getFirstMappedPort()
-            database = container!!.getDatabaseName()
-            username = container!!.getUsername()
-            password = container!!.getPassword()
-            driverClassName = container!!.getDriverClassName()
-            jdbcUrl = container!!.getJdbcUrl()
-        }
+        var database: String = container.databaseName
+        var driverClassName: String = container.driverClassName
+        var host: String = container.host
+        var jdbcUrl: String = container.getJdbcUrl()
+        var password: String = container.password
+        var port: Int = container.firstMappedPort
+        var username: String = container.username
     }
 }

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/db/jdbc/TestJdbcUtils.kt

@@ -33,7 +33,7 @@ import org.testcontainers.containers.PostgreSQLContainer
 import org.testcontainers.utility.MountableFile
 
 internal class TestJdbcUtils {
-    private var dbName: String? = null
+    private var dbName: String = "dummy"
     private lateinit var dataSource: DataSource
     @BeforeEach
     @Throws(Exception::class)
@@ -216,11 +216,11 @@ internal class TestJdbcUtils {
         val config =
             Jsons.jsonNode(
                 ImmutableMap.builder<Any, Any?>()
-                    .put("host", PSQL_DB!!.host)
-                    .put("port", PSQL_DB!!.firstMappedPort)
+                    .put("host", PSQL_DB.host)
+                    .put("port", PSQL_DB.firstMappedPort)
                     .put("database", dbName)
-                    .put("username", PSQL_DB!!.username)
-                    .put("password", PSQL_DB!!.password)
+                    .put("username", PSQL_DB.username)
+                    .put("password", PSQL_DB.password)
                     .put(
                         "ssl_mode",
                         ImmutableMap.builder<Any, Any>()
@@ -242,11 +242,11 @@ internal class TestJdbcUtils {
         val config =
             Jsons.jsonNode(
                 ImmutableMap.builder<Any, Any?>()
-                    .put("host", PSQL_DB!!.host)
-                    .put("port", PSQL_DB!!.firstMappedPort)
+                    .put("host", PSQL_DB.host)
+                    .put("port", PSQL_DB.firstMappedPort)
                     .put("database", dbName)
-                    .put("username", PSQL_DB!!.username)
-                    .put("password", PSQL_DB!!.password)
+                    .put("username", PSQL_DB.username)
+                    .put("password", PSQL_DB.password)
                     .put(
                         "ssl_mode",
                         ImmutableMap.builder<Any, Any>().put("mode", "disable").build()

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/integrations/base/ssh/SshTunnelTest.kt

@@ -28,8 +28,8 @@ internal class SshTunnelTest {
     @ParameterizedTest
     @ValueSource(strings = [HOST_PORT_CONFIG, URL_CONFIG_WITH_PORT, URL_CONFIG_NO_PORT])
     @Throws(Exception::class)
-    fun testConfigInTunnel(configString: String?) {
-        val config = ObjectMapper().readTree(String.format(configString!!, SSH_RSA_PRIVATE_KEY))
+    fun testConfigInTunnel(configString: String) {
+        val config = ObjectMapper().readTree(String.format(configString, SSH_RSA_PRIVATE_KEY))
         val endPointURL = Jsons.getStringOrNull(config, "endpoint")
         val sshTunnel: SshTunnel =
             object :

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/integrations/destination/buffered_stream_consumer/BufferedStreamConsumerTest.kt

@@ -29,21 +29,16 @@ import org.mockito.kotlin.any
 import org.mockito.kotlin.mock
 
 class BufferedStreamConsumerTest {
-    private lateinit var consumer: BufferedStreamConsumer
-    private lateinit var onStart: OnStartFunction
-    private lateinit var recordWriter: RecordWriter<AirbyteRecordMessage>
-    private lateinit var onClose: OnCloseFunction
-    private lateinit var isValidRecord: CheckedFunction<JsonNode?, Boolean?, Exception?>
-    private lateinit var outputRecordCollector: Consumer<AirbyteMessage>
+    private var consumer: BufferedStreamConsumer = mock()
+    private var onStart: OnStartFunction = mock()
+    private var recordWriter: RecordWriter<AirbyteRecordMessage> = mock()
+    private var onClose: OnCloseFunction = mock()
+    private var isValidRecord: CheckedFunction<JsonNode?, Boolean?, Exception?> = mock()
+    private var outputRecordCollector: Consumer<AirbyteMessage> = mock()
 
     @BeforeEach
     @Throws(Exception::class)
     fun setup() {
-        onStart = mock()
-        recordWriter = mock()
-        onClose = mock()
-        isValidRecord = mock()
-        outputRecordCollector = mock()
         consumer =
             BufferedStreamConsumer(
                 outputRecordCollector,

airbyte-cdk/java/airbyte-cdk/core/src/testFixtures/kotlin/io/airbyte/cdk/extensions/LoggingInvocationInterceptor.kt

@@ -35,12 +35,12 @@ import org.slf4j.LoggerFactory
 class LoggingInvocationInterceptor : InvocationInterceptor {
     private class LoggingInvocationInterceptorHandler : InvocationHandler {
         @Throws(Throwable::class)
-        override fun invoke(proxy: Any?, method: Method?, args: Array<Any?>?): Any? {
+        override fun invoke(proxy: Any, method: Method, args: Array<Any>): Any? {
             if (
                 LoggingInvocationInterceptor::class
                     .java
                     .getDeclaredMethod(
-                        method!!.name,
+                        method.name,
                         InvocationInterceptor.Invocation::class.java,
                         ReflectiveInvocationContext::class.java,
                         ExtensionContext::class.java
@@ -164,15 +164,15 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
         companion object {
             private val methodPattern: Pattern? = Pattern.compile("intercept(.*)Method")
 
-            private val PATTERN: Pattern? =
+            private val PATTERN: Pattern =
                 Pattern.compile(
                     "([1-9]\\d*) *((?:[nμm]?s)|m|h|d)?",
                     Pattern.CASE_INSENSITIVE or Pattern.UNICODE_CASE
                 )
-            private val UNITS_BY_ABBREVIATION: MutableMap<String?, TimeUnit?>?
+            private val UNITS_BY_ABBREVIATION: MutableMap<String, TimeUnit>
 
             init {
-                val unitsByAbbreviation: MutableMap<String?, TimeUnit?> = HashMap()
+                val unitsByAbbreviation: MutableMap<String, TimeUnit> = HashMap()
                 unitsByAbbreviation["ns"] = TimeUnit.NANOSECONDS
                 unitsByAbbreviation["μs"] = TimeUnit.MICROSECONDS
                 unitsByAbbreviation["ms"] = TimeUnit.MILLISECONDS
@@ -184,15 +184,15 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
             }
 
             @Throws(DateTimeParseException::class)
-            fun parseDuration(text: String?): Duration? {
-                val matcher = PATTERN!!.matcher(text!!.trim { it <= ' ' })
+            fun parseDuration(text: String): Duration {
+                val matcher = PATTERN.matcher(text.trim { it <= ' ' })
                 if (matcher.matches()) {
                     val value = matcher.group(1).toLong()
                     val unitAbbreviation = matcher.group(2)
                     val unit =
                         if (unitAbbreviation == null) TimeUnit.SECONDS
-                        else UNITS_BY_ABBREVIATION!![unitAbbreviation.lowercase()]
-                    return Duration.ofSeconds(unit!!.toSeconds(value))
+                        else UNITS_BY_ABBREVIATION.getValue(unitAbbreviation.lowercase())
+                    return Duration.ofSeconds(unit.toSeconds(value))
                 }
                 throw DateTimeParseException(
                     "Timeout duration is not in the expected format (<number> [ns|μs|ms|s|m|h|d])",
@@ -282,11 +282,11 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
 
     @Throws(Throwable::class)
     override fun interceptTestMethod(
-        invocation: InvocationInterceptor.Invocation<Void?>?,
-        invocationContext: ReflectiveInvocationContext<Method?>?,
-        extensionContext: ExtensionContext?
+        invocation: InvocationInterceptor.Invocation<Void>,
+        invocationContext: ReflectiveInvocationContext<Method>,
+        extensionContext: ExtensionContext
     ) {
-        if (!Modifier.isPublic(invocationContext!!.executable!!.modifiers)) {
+        if (!Modifier.isPublic(invocationContext.executable!!.modifiers)) {
             LOGGER!!.warn(
                 "Junit method {}.{} is not declared as public",
                 invocationContext.executable!!.declaringClass.canonicalName,

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

@@ -51,10 +51,10 @@ class SshBastionContainer : AutoCloseable {
     }
 
     @Throws(IOException::class, InterruptedException::class)
-    fun getTunnelMethod(tunnelMethod: SshTunnel.TunnelMethod?, innerAddress: Boolean): JsonNode? {
+    fun getTunnelMethod(tunnelMethod: SshTunnel.TunnelMethod, innerAddress: Boolean): JsonNode? {
         val containerAddress =
-            if (innerAddress) getInnerContainerAddress(container)
-            else getOuterContainerAddress(container)
+            if (innerAddress) getInnerContainerAddress(container!!)
+            else getOuterContainerAddress(container!!)
         return Jsons.jsonNode(
             ImmutableMap.builder<Any?, Any?>()
                 .put("tunnel_host", Objects.requireNonNull(containerAddress!!.left))
@@ -78,45 +78,45 @@ class SshBastionContainer : AutoCloseable {
 
     @Throws(IOException::class, InterruptedException::class)
     fun getTunnelConfig(
-        tunnelMethod: SshTunnel.TunnelMethod?,
-        builderWithSchema: ImmutableMap.Builder<Any?, Any?>?,
+        tunnelMethod: SshTunnel.TunnelMethod,
+        builderWithSchema: ImmutableMap.Builder<Any, Any>,
         innerAddress: Boolean
     ): JsonNode? {
         return Jsons.jsonNode(
-            builderWithSchema!!
+            builderWithSchema
                 .put("tunnel_method", getTunnelMethod(tunnelMethod, innerAddress))
                 .build()
         )
     }
 
-    fun getBasicDbConfigBuider(db: JdbcDatabaseContainer<*>?): ImmutableMap.Builder<Any?, Any?>? {
-        return getBasicDbConfigBuider(db, db!!.databaseName)
+    fun getBasicDbConfigBuider(db: JdbcDatabaseContainer<*>): ImmutableMap.Builder<Any, Any> {
+        return getBasicDbConfigBuider(db, db.databaseName)
     }
 
     fun getBasicDbConfigBuider(
-        db: JdbcDatabaseContainer<*>?,
-        schemas: MutableList<String?>?
-    ): ImmutableMap.Builder<Any?, Any?>? {
-        return getBasicDbConfigBuider(db, db!!.databaseName)!!.put("schemas", schemas)
+        db: JdbcDatabaseContainer<*>,
+        schemas: MutableList<String>
+    ): ImmutableMap.Builder<Any, Any> {
+        return getBasicDbConfigBuider(db, db.databaseName).put("schemas", schemas)
     }
 
     fun getBasicDbConfigBuider(
-        db: JdbcDatabaseContainer<*>?,
-        schemaName: String?
-    ): ImmutableMap.Builder<Any?, Any?>? {
-        return ImmutableMap.builder<Any?, Any?>()
+        db: JdbcDatabaseContainer<*>,
+        schemaName: String
+    ): ImmutableMap.Builder<Any, Any> {
+        return ImmutableMap.builder<Any, Any>()
             .put("host", Objects.requireNonNull(HostPortResolver.resolveHost(db)))
-            .put("username", db!!.username)
+            .put("username", db.username)
             .put("password", db.password)
             .put("port", HostPortResolver.resolvePort(db))
             .put("database", schemaName)
             .put("ssl", false)
     }
 
-    fun stopAndCloseContainers(db: JdbcDatabaseContainer<*>?) {
+    fun stopAndCloseContainers(db: JdbcDatabaseContainer<*>) {
         container!!.stop()
         container!!.close()
-        db!!.stop()
+        db.stop()
         db.close()
     }
 
@@ -131,8 +131,8 @@ class SshBastionContainer : AutoCloseable {
     companion object {
         private val factory: SshBastionContainerFactory? = SshBastionContainerFactory()
 
-        private val SSH_USER: String? = "sshuser"
-        private val SSH_PASSWORD: String? = "secret"
+        private val SSH_USER: String = "sshuser"
+        private val SSH_PASSWORD: String = "secret"
 
         @JvmStatic
         /**
@@ -142,13 +142,9 @@ class SshBastionContainer : AutoCloseable {
          * @param container container
          * @return a pair of host and port
          */
-        fun getInnerContainerAddress(container: Container<*>?): ImmutablePair<String?, Int?>? {
+        fun getInnerContainerAddress(container: Container<*>): ImmutablePair<String, Int> {
             return ImmutablePair.of(
-                container!!
-                    .containerInfo
-                    .networkSettings
-                    .networks
-                    .entries
+                container.containerInfo.networkSettings.networks.entries
                     .stream()
                     .findFirst()
                     .get()
@@ -166,8 +162,8 @@ class SshBastionContainer : AutoCloseable {
          * @param container container
          * @return a pair of host and port
          */
-        fun getOuterContainerAddress(container: Container<*>?): ImmutablePair<String?, Int?>? {
-            return ImmutablePair.of(container!!.host, container.firstMappedPort)
+        fun getOuterContainerAddress(container: Container<*>): ImmutablePair<String, Int> {
+            return ImmutablePair.of(container.host, container.firstMappedPort)
         }
     }
 }