Merge branch 'master' into btkcodedev/builder-contribute/source-buzzs…

…prout

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.64.2
+current_version = 0.64.4
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(\-[a-z]+)?

.github/workflows/airbyte-ci-release.yml

@@ -54,7 +54,7 @@ jobs:
         working-directory: airbyte-ci/connectors/pipelines/
         run: poetry run poe build-release-binary ${{ env.BINARY_FILE_NAME }}
 
-      - uses: actions/upload-artifact@v2
+      - uses: actions/upload-artifact@v3
         with:
           name: airbyte-ci-${{ matrix.os }}-${{ steps.get_short_sha.outputs.sha }}
           path: airbyte-ci/connectors/pipelines/dist/${{ env.BINARY_FILE_NAME }}

.github/workflows/community_ci.yml

@@ -76,60 +76,6 @@ jobs:
           sentry_dsn: ${{ secrets.SENTRY_AIRBYTE_CI_DSN }}
           subcommand: "format check all"
           is_fork: "true"
-
-  check-review-requirements:
-    name: Check if a review is required from Connector teams on fork
-    if: github.event.pull_request.head.repo.fork == true
-    environment: community-ci-auto
-    runs-on: community-tooling-test-small
-    needs: fail_on_protected_path_changes
-    timeout-minutes: 10
-    env:
-      MAIN_BRANCH_NAME: "master"
-    permissions:
-      pull-requests: write
-    steps:
-      # This checkouts a fork which can contain untrusted code
-      # It's deemed safe as the review required check is not executing any checked out code
-      - name: Checkout fork
-        uses: actions/checkout@v4
-        with:
-          repository: ${{ github.event.pull_request.head.repo.full_name }}
-          ref: ${{ github.event.pull_request.head.sha }}
-          fetch-depth: 1
-      # This will sync the .github folder of the main repo with the fork
-      # This allows us to use up to date actions and CI logic from the main repo
-      - name: Pull .github folder from main repository
-        id: pull_github_folder
-        run: |
-          git remote add main https://github.com/airbytehq/airbyte.git
-          git fetch main ${MAIN_BRANCH_NAME}
-          git checkout main/${MAIN_BRANCH_NAME} -- .github
-          git checkout main/${MAIN_BRANCH_NAME} -- airbyte-ci
-      - name: Install Python
-        uses: actions/setup-python@v4
-        with:
-          python-version: "3.10"
-      - name: Install ci-connector-ops package
-        run: |
-          pip install pipx
-          pipx ensurepath
-          pipx install airbyte-ci/connectors/connector_ops
-      - name: Write review requirements file
-        id: write-review-requirements-file
-        run: write-review-requirements-file >> $GITHUB_OUTPUT
-      - name: Get mandatory reviewers
-        id: get-mandatory-reviewers
-        run: print-mandatory-reviewers >> $GITHUB_OUTPUT
-      - name: Check if the review requirements are met
-        if: steps.write-review-requirements-file.outputs.CREATED_REQUIREMENTS_FILE == 'true'
-        uses: Automattic/action-required-review@v3
-        with:
-          status: ${{ steps.get-mandatory-reviewers.outputs.MANDATORY_REVIEWERS  }}
-          token: ${{ secrets.OCTAVIA_4_ROOT_ACCESS }}
-          request-reviews: true
-          requirements-file: .github/connector_org_review_requirements.yaml
-
   connectors_early_ci:
     name: Run connectors early CI on fork
     if: github.event.pull_request.head.repo.fork == true

.gitignore

@@ -10,6 +10,7 @@ out
 .project
 .settings
 .vscode
+.kotlin
 **/gmon.out
 static_checker_reports/
 

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

@@ -5,6 +5,7 @@ import io.airbyte.cdk.command.ConnectorCommandLinePropertySource
 import io.airbyte.cdk.command.MetadataYamlPropertySource
 import io.micronaut.configuration.picocli.MicronautFactory
 import io.micronaut.context.ApplicationContext
+import io.micronaut.context.RuntimeBeanDefinition
 import io.micronaut.context.env.CommandLinePropertySource
 import io.micronaut.context.env.Environment
 import io.micronaut.core.cli.CommandLine as MicronautCommandLine
@@ -17,8 +18,11 @@ import picocli.CommandLine.Model.UsageMessageSpec
 
 /** Source connector entry point. */
 class AirbyteSourceRunner(
+    /** CLI args. */
     args: Array<out String>,
-) : AirbyteConnectorRunner("source", args) {
+    /** Micronaut bean definition overrides, used only for tests. */
+    vararg testBeanDefinitions: RuntimeBeanDefinition<*>,
+) : AirbyteConnectorRunner("source", args, testBeanDefinitions) {
     companion object {
         @JvmStatic
         fun run(vararg args: String) {
@@ -29,8 +33,11 @@ class AirbyteSourceRunner(
 
 /** Destination connector entry point. */
 class AirbyteDestinationRunner(
+    /** CLI args. */
     args: Array<out String>,
-) : AirbyteConnectorRunner("destination", args) {
+    /** Micronaut bean definition overrides, used only for tests. */
+    vararg testBeanDefinitions: RuntimeBeanDefinition<*>,
+) : AirbyteConnectorRunner("destination", args, testBeanDefinitions) {
     companion object {
         @JvmStatic
         fun run(vararg args: String) {
@@ -46,6 +53,7 @@ class AirbyteDestinationRunner(
 sealed class AirbyteConnectorRunner(
     val connectorType: String,
     val args: Array<out String>,
+    val testBeanDefinitions: Array<out RuntimeBeanDefinition<*>>,
 ) {
     val envs: Array<String> = arrayOf(Environment.CLI, connectorType)
 
@@ -65,11 +73,12 @@ sealed class AirbyteConnectorRunner(
                     commandLinePropertySource,
                     MetadataYamlPropertySource(),
                 )
+                .beanDefinitions(*testBeanDefinitions)
                 .start()
         val isTest: Boolean = ctx.environment.activeNames.contains(Environment.TEST)
         val picocliFactory: CommandLine.IFactory = MicronautFactory(ctx)
         val picocliCommandLine: CommandLine =
-            picocliCommandLineFactory.build<AirbyteConnectorRunnable>(picocliFactory, isTest)
+            picocliCommandLineFactory.build<AirbyteConnectorRunnable>(picocliFactory)
         val exitCode: Int = picocliCommandLine.execute(*args)
         if (!isTest) {
             // Required by the platform, otherwise syncs may hang.
@@ -82,10 +91,7 @@ sealed class AirbyteConnectorRunner(
 class PicocliCommandLineFactory(
     val runner: AirbyteConnectorRunner,
 ) {
-    inline fun <reified R : Runnable> build(
-        factory: CommandLine.IFactory,
-        isTest: Boolean,
-    ): CommandLine {
+    inline fun <reified R : Runnable> build(factory: CommandLine.IFactory): CommandLine {
         val commandSpec: CommandLine.Model.CommandSpec =
             CommandLine.Model.CommandSpec.wrapWithoutInspection(R::class.java, factory)
                 .name("airbyte-${runner.connectorType}-connector")
@@ -95,10 +101,6 @@ class PicocliCommandLineFactory(
                 .addOption(config)
                 .addOption(catalog)
                 .addOption(state)
-
-        if (isTest) {
-            commandSpec.addOption(output)
-        }
         return CommandLine(commandSpec, factory)
     }
 
@@ -168,10 +170,4 @@ class PicocliCommandLineFactory(
             "path to the json-encoded state file",
             "Required by the following commands: read",
         )
-    val output: OptionSpec =
-        fileOption(
-            "output",
-            "path to the output file",
-            "When present, the connector writes to this file instead of stdout",
-        )
 }

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

@@ -23,7 +23,6 @@ class ConnectorCommandLinePropertySource(
 const val CONNECTOR_CONFIG_PREFIX: String = "airbyte.connector.config"
 const val CONNECTOR_CATALOG_PREFIX: String = "airbyte.connector.catalog"
 const val CONNECTOR_STATE_PREFIX: String = "airbyte.connector.state"
-const val CONNECTOR_OUTPUT_FILE = "airbyte.connector.output.file"
 
 private fun resolveValues(
     commandLine: CommandLine,
@@ -39,7 +38,6 @@ private fun resolveValues(
     }
     val values: MutableMap<String, Any> = mutableMapOf()
     values[Operation.PROPERTY] = ops.first()
-    commandLine.optionValue("output")?.let { values[CONNECTOR_OUTPUT_FILE] = it }
     for ((cliOptionKey, prefix) in
         mapOf(
             "config" to CONNECTOR_CONFIG_PREFIX,

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

@@ -25,9 +25,7 @@ import io.micronaut.context.annotation.Value
 import io.micronaut.context.env.Environment
 import jakarta.inject.Singleton
 import java.io.ByteArrayOutputStream
-import java.io.FileOutputStream
 import java.io.PrintStream
-import java.nio.file.Path
 import java.time.Clock
 import java.time.Instant
 import java.util.concurrent.ConcurrentHashMap
@@ -104,9 +102,6 @@ interface OutputConsumer : Consumer<AirbyteMessage>, AutoCloseable {
 /** Configuration properties prefix for [StdoutOutputConsumer]. */
 const val CONNECTOR_OUTPUT_PREFIX = "airbyte.connector.output"
 
-// Used for integration tests.
-const val CONNECTOR_OUTPUT_FILE = "$CONNECTOR_OUTPUT_PREFIX.file"
-
 /** Default implementation of [OutputConsumer]. */
 @Singleton
 @Secondary
@@ -293,10 +288,4 @@ private class RecordTemplate(
 private class PrintStreamFactory {
 
     @Singleton @Requires(notEnv = [Environment.TEST]) fun stdout(): PrintStream = System.out
-
-    @Singleton
-    @Requires(env = [Environment.TEST])
-    @Requires(property = CONNECTOR_OUTPUT_FILE)
-    fun file(@Value("\${$CONNECTOR_OUTPUT_FILE}") filePath: Path): PrintStream =
-        PrintStream(FileOutputStream(filePath.toFile()), false, Charsets.UTF_8)
 }

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

@@ -5,6 +5,7 @@ import io.airbyte.cdk.Operation
 import io.airbyte.cdk.command.ConfigurationJsonObjectSupplier
 import io.airbyte.cdk.output.OutputConsumer
 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.Singleton
@@ -15,13 +16,24 @@ import java.net.URI
 class SpecOperation(
     @Value("\${airbyte.connector.metadata.documentation-url}") val documentationUrl: String,
     val configJsonObjectSupplier: ConfigurationJsonObjectSupplier<*>,
+    val extendSpecification: SpecificationExtender,
     val outputConsumer: OutputConsumer,
 ) : Operation {
     override fun execute() {
-        outputConsumer.accept(
+        val spec =
             ConnectorSpecification()
                 .withDocumentationUrl(URI.create(documentationUrl))
-                .withConnectionSpecification(configJsonObjectSupplier.jsonSchema),
-        )
+                .withConnectionSpecification(configJsonObjectSupplier.jsonSchema)
+        outputConsumer.accept(extendSpecification(spec))
+    }
+}
+
+interface SpecificationExtender : (ConnectorSpecification) -> ConnectorSpecification
+
+@Singleton
+@DefaultImplementation
+class IdentitySpecificationExtender : SpecificationExtender {
+    override fun invoke(specification: ConnectorSpecification): ConnectorSpecification {
+        return specification
     }
 }

airbyte-cdk/bulk/core/base/src/testFixtures/kotlin/io/airbyte/cdk/command/CliRunnable.kt

@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.command
+
+import io.airbyte.cdk.output.BufferingOutputConsumer
+import io.airbyte.protocol.models.v0.AirbyteMessage
+
+/** Convenience object for return values in [CliRunner]. */
+data class CliRunnable(
+    val runnable: Runnable,
+    val results: BufferingOutputConsumer,
+) {
+
+    /** Decorates the [BufferingOutputConsumer] with a callback, which should return quickly. */
+    fun withCallback(nonBlockingFn: (AirbyteMessage) -> Unit): CliRunnable {
+        results.callback = nonBlockingFn
+        return this
+    }
+
+    /** Runs the [Runnable]. */
+    fun run(): BufferingOutputConsumer {
+        runnable.run()
+        return results
+    }
+}