allow connectors to be written in kotlin

Author: stephane-airbyte

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

@@ -3,19 +3,6 @@ plugins {
     id "java-library"
 }
 
-java {
-    // TODO: rewrite code to avoid javac wornings in the first place
-    compileJava {
-        options.compilerArgs += "-Xlint:-varargs,-try,-deprecation,-unchecked,-this-escape"
-    }
-    compileTestJava {
-        options.compilerArgs += "-Xlint:-try"
-    }
-    compileTestFixturesJava {
-        options.compilerArgs += "-Xlint:-try"
-    }
-}
-compileKotlin.compilerOptions.allWarningsAsErrors = false
 compileTestFixturesKotlin.compilerOptions.allWarningsAsErrors = false
 compileTestKotlin.compilerOptions.allWarningsAsErrors = false
 

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/concurrency/CompletableFutures.kt

@@ -24,21 +24,18 @@ object CompletableFutures {
         val result = CompletableFuture<List<Either<out Exception, Result>>>()
         val size = futures.size
         val counter = AtomicInteger()
-        val results =
-            java.lang.reflect.Array.newInstance(Either::class.java, size)
-                as Array<Either<Exception, Result>>
+        val results = mutableListOf<Either<out Exception, Result>>()
         // attach a whenComplete to all futures
-        for (i in 0 until size) {
-            val currentIndex = i
-            futures[i].whenComplete { value: Result, exception: Throwable? ->
+        for (future in futures) {
+            future.whenComplete { value: Result, exception: Throwable? ->
                 // if exception is null, then the future completed successfully
                 // maybe synchronization is unnecessary here, but it's better to be safe
                 synchronized(results) {
                     if (exception == null) {
-                        results[currentIndex] = Either.right(value)
+                        results.add(Either.right(value))
                     } else {
                         if (exception is Exception) {
-                            results[currentIndex] = Either.left(exception)
+                            results.add(Either.left(exception))
                         } else {
                             // this should never happen
                             throw RuntimeException(
@@ -50,7 +47,7 @@ object CompletableFutures {
                 }
                 val completedCount = counter.incrementAndGet()
                 if (completedCount == size) {
-                    result.complete(Arrays.asList(*results))
+                    result.complete(results)
                 }
             }
         }

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/enums/Enums.kt

@@ -15,6 +15,7 @@ import java.util.stream.Collectors
 
 class Enums {
     companion object {
+        @Suppress("UNUSED_PARAMETER")
         inline fun <T1 : Enum<T1>, reified T2 : Enum<T2>> convertTo(ie: T1?, oe: Class<T2>): T2? {
             if (ie == null) {
                 return null

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/features/FeatureFlagHelper.kt

@@ -41,7 +41,7 @@ object FeatureFlagHelper {
                 try {
                     workspaceIds.add(UUID.fromString(id))
                 } catch (e: IllegalArgumentException) {
-                    log.warn("Malformed workspace id for {}: {}", context, id)
+                    log.warn { "Malformed workspace id for $context: $id" }
                 }
             }
         }

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/functional/Either.kt

@@ -29,14 +29,14 @@ class Either<Error, Result> private constructor(left: Error?, right: Result?) {
         return right != null
     }
 
-    override fun equals(o: Any?): Boolean {
-        if (this === o) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) {
             return true
         }
-        if (o == null || javaClass != o.javaClass) {
+        if (other == null || javaClass != other.javaClass) {
             return false
         }
-        val either = o as Either<*, *>
+        val either = other as Either<*, *>
         return left == either.left && right == either.right
     }
 
@@ -47,7 +47,7 @@ class Either<Error, Result> private constructor(left: Error?, right: Result?) {
     companion object {
         fun <Error, Result> left(error: Error): Either<Error, Result> {
             if (error == null) {
-                LOGGER.warn("Either.left called with a null!")
+                LOGGER.warn { "Either.left called with a null!" }
             }
             return Either(error!!, null)
         }

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/io/IOs.kt

@@ -83,7 +83,8 @@ object IOs {
             return emptyList<String>()
         }
 
-        ReversedLinesFileReader(file, Charsets.UTF_8).use { fileReader ->
+        ReversedLinesFileReader.Builder().setFile(file).setCharset(Charsets.UTF_8).get().use {
+            fileReader ->
             val lines: MutableList<String?> = ArrayList()
             var line = fileReader.readLine()
             while (line != null && lines.size < numLines) {

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/io/LineGobbler.kt

@@ -27,7 +27,7 @@ internal constructor(
     private val caller: String = GENERIC,
     private val containerLogMdcBuilder: MdcScope.Builder = MdcScope.Companion.DEFAULT_BUILDER
 ) : VoidCallable {
-    private val `is`: BufferedReader? = IOs.newBufferedReader(`is`)
+    private val `is`: BufferedReader = IOs.newBufferedReader(`is`)
 
     internal constructor(
         `is`: InputStream,
@@ -40,9 +40,9 @@ internal constructor(
     override fun voidCall() {
         MDC.setContextMap(mdc)
         try {
-            var line = `is`!!.readLine()
+            var line = `is`.readLine()
             while (line != null) {
-                containerLogMdcBuilder.build().use { mdcScope -> consumer.accept(line) }
+                containerLogMdcBuilder.build().use { consumer.accept(line) }
                 line = `is`.readLine()
             }
         } catch (i: IOException) {

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/json/JsonPaths.kt

@@ -259,7 +259,7 @@ object JsonPaths {
      * @param replacement
      * - a string value to replace the current value at the jsonPath
      */
-    fun replaceAtString(json: JsonNode, jsonPath: String, replacement: String): JsonNode? {
+    fun replaceAtString(json: JsonNode, jsonPath: String, replacement: String): JsonNode {
         return replaceAtJsonNode(json, jsonPath, Jsons.jsonNode(replacement))
     }
 
@@ -315,7 +315,7 @@ object JsonPaths {
         json: JsonNode,
         jsonPath: String,
         replacementFunction: BiFunction<JsonNode, String, JsonNode>
-    ): JsonNode? {
+    ): JsonNode {
         var clone = Jsons.clone(json)
         assertIsJsonPath(jsonPath)
         val foundPaths = getPaths(clone, jsonPath)

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/json/JsonSchemas.kt

@@ -226,9 +226,9 @@ object JsonSchemas {
                             consumer
                         )
                     } else {
-                        log.warn(
+                        log.warn {
                             "The array is missing an items field. The traversal is silently stopped. Current schema: $jsonSchemaNode"
-                        )
+                        }
                     }
                 }
                 OBJECT_TYPE -> {
@@ -247,9 +247,9 @@ object JsonSchemas {
                             traverseJsonSchemaInternal(arrayItem, path, consumer)
                         }
                     } else {
-                        log.warn(
+                        log.warn {
                             "The object is a properties key or a combo keyword. The traversal is silently stopped. Current schema: $jsonSchemaNode"
-                        )
+                        }
                     }
                 }
             }
@@ -331,14 +331,14 @@ object JsonSchemas {
     class FieldNameOrList private constructor(val fieldName: String?) {
         val isList: Boolean = fieldName == null
 
-        override fun equals(o: Any?): Boolean {
-            if (this === o) {
+        override fun equals(other: Any?): Boolean {
+            if (this === other) {
                 return true
             }
-            if (o !is FieldNameOrList) {
+            if (other !is FieldNameOrList) {
                 return false
             }
-            val that = o
+            val that = other
             return isList == that.isList && fieldName == that.fieldName
         }
 

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/json/Jsons.kt

@@ -221,7 +221,7 @@ object Jsons {
 
     @JvmStatic
     fun <T : Any> clone(o: T): T {
-        return deserialize(serialize(o), o::class.java) as T
+        return deserialize(serialize(o), o::class.java)
     }
 
     fun toBytes(jsonNode: JsonNode): ByteArray {
@@ -262,16 +262,16 @@ object Jsons {
     }
 
     fun navigateTo(node: JsonNode, keys: List<String?>): JsonNode {
-        var node = node
+        var targetNode = node
         for (key in keys) {
-            node = node[key]
+            targetNode = targetNode[key]
         }
-        return node
+        return targetNode
     }
 
     fun replaceNestedValue(json: JsonNode, keys: List<String?>, replacement: JsonNode?) {
         replaceNested(json, keys) { node: ObjectNode, finalKey: String? ->
-            node.put(finalKey, replacement)
+            node.replace(finalKey, replacement)
         }
     }
 
@@ -302,16 +302,16 @@ object Jsons {
     }
 
     fun getOptional(json: JsonNode?, keys: List<String>): Optional<JsonNode> {
-        var json = json
+        var retVal = json
         for (key in keys) {
-            if (json == null) {
+            if (retVal == null) {
                 return Optional.empty()
             }
 
-            json = json[key]
+            retVal = retVal[key]
         }
 
-        return Optional.ofNullable(json)
+        return Optional.ofNullable(retVal)
     }
 
     fun getStringOrNull(json: JsonNode?, vararg keys: String): String? {
@@ -419,21 +419,21 @@ object Jsons {
      * the class name can at least help narrow down the problem, without leaking
      * potentially-sensitive information. </snip...>
      */
-    private fun <T : Any> handleDeserThrowable(t: Throwable): Optional<T> {
+    private fun <T : Any> handleDeserThrowable(throwable: Throwable): Optional<T> {
         // Manually build the stacktrace, excluding the top-level exception object
         // so that we don't accidentally include the exception message.
         // Otherwise we could just do ExceptionUtils.getStackTrace(t).
-        var t: Throwable? = t
+        var t: Throwable = throwable
         val sb = StringBuilder()
-        sb.append(t!!.javaClass)
+        sb.append(t.javaClass)
         for (traceElement in t.stackTrace) {
             sb.append("\n\tat ")
             sb.append(traceElement.toString())
         }
-        while (t!!.cause != null) {
-            t = t.cause
+        while (t.cause != null) {
+            t = t.cause!!
             sb.append("\nCaused by ")
-            sb.append(t!!.javaClass)
+            sb.append(t.javaClass)
             for (traceElement in t.stackTrace) {
                 sb.append("\n\tat ")
                 sb.append(traceElement.toString())

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/stream/StreamStatusUtils.kt

@@ -62,9 +62,7 @@ object StreamStatusUtils {
         airbyteStream: AutoCloseableIterator<AirbyteMessage>,
         statusEmitter: Optional<Consumer<AirbyteStreamStatusHolder>>
     ) {
-        if (airbyteStream is AirbyteStreamAware) {
-            emitRunningStreamStatus(airbyteStream as AirbyteStreamAware, statusEmitter)
-        }
+        emitRunningStreamStatus(airbyteStream as AirbyteStreamAware, statusEmitter)
     }
 
     /**
@@ -90,7 +88,7 @@ object StreamStatusUtils {
         airbyteStream: Optional<AirbyteStreamNameNamespacePair>,
         statusEmitter: Optional<Consumer<AirbyteStreamStatusHolder>>
     ) {
-        airbyteStream!!.ifPresent { s: AirbyteStreamNameNamespacePair? ->
+        airbyteStream.ifPresent { s: AirbyteStreamNameNamespacePair? ->
             LOGGER.debug("RUNNING -> {}", s)
             emitStreamStatus(
                 s,
@@ -110,9 +108,7 @@ object StreamStatusUtils {
         airbyteStream: AutoCloseableIterator<AirbyteMessage>,
         statusEmitter: Optional<Consumer<AirbyteStreamStatusHolder>>
     ) {
-        if (airbyteStream is AirbyteStreamAware) {
-            emitStartStreamStatus(airbyteStream as AirbyteStreamAware, statusEmitter)
-        }
+        emitStartStreamStatus(airbyteStream as AirbyteStreamAware, statusEmitter)
     }
 
     /**
@@ -138,7 +134,7 @@ object StreamStatusUtils {
         airbyteStream: Optional<AirbyteStreamNameNamespacePair>,
         statusEmitter: Optional<Consumer<AirbyteStreamStatusHolder>>
     ) {
-        airbyteStream!!.ifPresent { s: AirbyteStreamNameNamespacePair? ->
+        airbyteStream.ifPresent { s: AirbyteStreamNameNamespacePair? ->
             LOGGER.debug("STARTING -> {}", s)
             emitStreamStatus(
                 s,
@@ -158,9 +154,7 @@ object StreamStatusUtils {
         airbyteStream: AutoCloseableIterator<AirbyteMessage>,
         statusEmitter: Optional<Consumer<AirbyteStreamStatusHolder>>
     ) {
-        if (airbyteStream is AirbyteStreamAware) {
-            emitCompleteStreamStatus(airbyteStream as AirbyteStreamAware, statusEmitter)
-        }
+        emitCompleteStreamStatus(airbyteStream as AirbyteStreamAware, statusEmitter)
     }
 
     /**
@@ -186,7 +180,7 @@ object StreamStatusUtils {
         airbyteStream: Optional<AirbyteStreamNameNamespacePair>,
         statusEmitter: Optional<Consumer<AirbyteStreamStatusHolder>>
     ) {
-        airbyteStream!!.ifPresent { s: AirbyteStreamNameNamespacePair? ->
+        airbyteStream.ifPresent { s: AirbyteStreamNameNamespacePair? ->
             LOGGER.debug("COMPLETE -> {}", s)
             emitStreamStatus(
                 s,
@@ -206,9 +200,7 @@ object StreamStatusUtils {
         airbyteStream: AutoCloseableIterator<AirbyteMessage>,
         statusEmitter: Optional<Consumer<AirbyteStreamStatusHolder>>
     ) {
-        if (airbyteStream is AirbyteStreamAware) {
-            emitIncompleteStreamStatus(airbyteStream as AirbyteStreamAware, statusEmitter)
-        }
+        emitIncompleteStreamStatus(airbyteStream as AirbyteStreamAware, statusEmitter)
     }
 
     /**
@@ -234,7 +226,7 @@ object StreamStatusUtils {
         airbyteStream: Optional<AirbyteStreamNameNamespacePair>,
         statusEmitter: Optional<Consumer<AirbyteStreamStatusHolder>>
     ) {
-        airbyteStream!!.ifPresent { s: AirbyteStreamNameNamespacePair? ->
+        airbyteStream.ifPresent { s: AirbyteStreamNameNamespacePair? ->
             LOGGER.debug("INCOMPLETE -> {}", s)
             emitStreamStatus(
                 s,

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/util/CompositeIterator.kt

@@ -108,7 +108,7 @@ internal constructor(
     private fun emitStartStreamStatus(
         airbyteStream: Optional<AirbyteStreamNameNamespacePair>
     ): Boolean {
-        if (airbyteStream!!.isPresent && !seenIterators.contains(airbyteStream)) {
+        if (airbyteStream.isPresent && !seenIterators.contains(airbyteStream)) {
             seenIterators.add(airbyteStream)
             StreamStatusUtils.emitStartStreamStatus(airbyteStream, airbyteStreamStatusConsumer)
             return true
@@ -136,12 +136,7 @@ internal constructor(
     }
 
     override val airbyteStream: Optional<AirbyteStreamNameNamespacePair>
-        get() =
-            if (currentIterator() is AirbyteStreamAware) {
-                AirbyteStreamAware::class.java.cast(currentIterator()).airbyteStream
-            } else {
-                Optional.empty()
-            }
+        get() = AirbyteStreamAware::class.java.cast(currentIterator()).airbyteStream
 
     private fun assertHasNotClosed() {
         Preconditions.checkState(!hasClosed)

airbyte-cdk/java/airbyte-cdk/dependencies/src/main/kotlin/io/airbyte/commons/version/Version.kt

@@ -138,14 +138,14 @@ open class Version {
             '}'
     }
 
-    override fun equals(o: Any?): Boolean {
-        if (this === o) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) {
             return true
         }
-        if (o == null || javaClass != o.javaClass) {
+        if (other == null || javaClass != other.javaClass) {
             return false
         }
-        val that = o as Version
+        val that = other as Version
         return version == that.version &&
             major == that.major &&
             minor == that.minor &&