named root consumer

Author: johnny-schmidt

airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/AirbyteConnectorRunnable.kt

@@ -7,6 +7,7 @@ import io.airbyte.cdk.output.OutputConsumer
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.micronaut.context.annotation.Value
 import jakarta.inject.Inject
+import jakarta.inject.Named
 import jakarta.inject.Provider
 
 private val log = KotlinLogging.logger {}
@@ -18,7 +19,7 @@ class AirbyteConnectorRunnable : Runnable {
 
     @Inject lateinit var operationProvider: Provider<Operation>
 
-    @Inject lateinit var outputConsumer: OutputConsumer
+    @Inject @Named("stdoutOutputConsumer") lateinit var outputConsumer: OutputConsumer
 
     @Inject lateinit var exceptionHandler: ExceptionHandler
 

airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/output/OutputConsumer.kt

@@ -23,6 +23,7 @@ import io.micronaut.context.annotation.Requires
 import io.micronaut.context.annotation.Secondary
 import io.micronaut.context.annotation.Value
 import io.micronaut.context.env.Environment
+import jakarta.inject.Named
 import jakarta.inject.Singleton
 import java.io.ByteArrayOutputStream
 import java.io.PrintStream
@@ -119,6 +120,7 @@ const val CONNECTOR_OUTPUT_PREFIX = "airbyte.connector.output"
 /** Default implementation of [OutputConsumer]. */
 @Singleton
 @Secondary
+@Named("stdoutOutputConsumer")
 open class StdoutOutputConsumer(
     val stdout: PrintStream,
     val clock: Clock,

airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/spec/SpecOperation.kt

@@ -8,6 +8,7 @@ import io.airbyte.protocol.models.v0.ConnectorSpecification
 import io.micronaut.context.annotation.DefaultImplementation
 import io.micronaut.context.annotation.Requires
 import io.micronaut.context.annotation.Value
+import jakarta.inject.Named
 import jakarta.inject.Singleton
 import java.net.URI
 
@@ -17,7 +18,7 @@ class SpecOperation(
     @Value("\${airbyte.connector.metadata.documentation-url}") val documentationUrl: String,
     val configJsonObjectSupplier: ConfigurationSpecificationSupplier<*>,
     val extendSpecification: SpecificationExtender,
-    val outputConsumer: OutputConsumer,
+    @Named("stdoutOutputConsumer") val outputConsumer: OutputConsumer,
 ) : Operation {
     override fun execute() {
         val spec =

airbyte-cdk/bulk/core/extract/src/main/kotlin/io/airbyte/cdk/discover/DiscoverOperation.kt

@@ -8,6 +8,7 @@ import io.airbyte.protocol.models.v0.AirbyteCatalog
 import io.airbyte.protocol.models.v0.AirbyteStream
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.micronaut.context.annotation.Requires
+import jakarta.inject.Named
 import jakarta.inject.Singleton
 
 @Singleton
@@ -17,7 +18,7 @@ class DiscoverOperation(
     val config: SourceConfiguration,
     val metadataQuerierFactory: MetadataQuerier.Factory<SourceConfiguration>,
     val airbyteStreamFactory: AirbyteStreamFactory,
-    val outputConsumer: OutputConsumer,
+    @Named("stdoutOutputConsumer") val outputConsumer: OutputConsumer,
 ) : Operation {
     private val log = KotlinLogging.logger {}
 

airbyte-cdk/bulk/core/extract/src/main/kotlin/io/airbyte/cdk/read/ReadOperation.kt

@@ -14,6 +14,7 @@ import io.airbyte.cdk.util.ThreadRenamingCoroutineName
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.micronaut.context.annotation.Requires
+import jakarta.inject.Named
 import jakarta.inject.Singleton
 import kotlin.time.toKotlinDuration
 import kotlinx.coroutines.CoroutineScope