Merge latest changes from main branch

.github/workflows/community_ci.yml

@@ -172,7 +172,6 @@ jobs:
         uses: ./.github/actions/run-airbyte-ci
         with:
           context: "pull_request"
-          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_4 }}
           docker_hub_password: ${{ secrets.DOCKER_HUB_PASSWORD }}
           docker_hub_username: ${{ secrets.DOCKER_HUB_USERNAME }}
           gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}
@@ -237,7 +236,6 @@ jobs:
         uses: ./.github/actions/run-airbyte-ci
         with:
           context: "pull_request"
-          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_4 }}
           docker_hub_password: ${{ secrets.DOCKER_HUB_PASSWORD }}
           docker_hub_username: ${{ secrets.DOCKER_HUB_USERNAME }}
           gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}

.github/workflows/connectors_up_to_date.yml

@@ -11,7 +11,7 @@ on:
     inputs:
       connectors-options:
         description: "Options to pass to the 'airbyte-ci connectors' command group."
-        default: "--concurrency=10 --language=python --language=low-code --language=manifest-only"
+        default: '--concurrency=10 --language=python --language=low-code --language=manifest-only --metadata-query="''-rc.'' not in data.dockerImageTag" --metadata-query="''source-declarative-manifest'' not in data.dockerRepository"'
       auto-merge:
         description: "Whether to auto-merge the PRs created by the action."
         default: "false"
@@ -25,8 +25,9 @@ jobs:
     steps:
       - name: Checkout Airbyte
         uses: actions/checkout@v4
-      - name: Run airbyte-ci connectors up-to-date
-        id: airbyte-ci-connectors-up-to-date
+      - name: Run airbyte-ci connectors up-to-date [WORKFLOW]
+        if: github.event_name == 'workflow_dispatch'
+        id: airbyte-ci-connectors-up-to-date-workflow-dispatch
         uses: ./.github/actions/run-airbyte-ci
         with:
           context: "master"
@@ -39,4 +40,20 @@ jobs:
           sentry_dsn: ${{ secrets.SENTRY_AIRBYTE_CI_DSN }}
           s3_build_cache_access_key_id: ${{ secrets.SELF_RUNNER_AWS_ACCESS_KEY_ID }}
           s3_build_cache_secret_key: ${{ secrets.SELF_RUNNER_AWS_SECRET_ACCESS_KEY }}
-          subcommand: "connectors  ${{ github.event.inputs.connectors-options || '--concurrency=10 --language=python --language=low-code --support-level=community --support-level=certified' }} up-to-date --ignore-connector=source-declarative-manifest --create-prs ${{ github.event.inputs.auto-merge == 'false' && '' || '--auto-merge' }}"
+          subcommand: "connectors ${{ github.event.inputs.connectors-options }} up-to-date  --create-prs ${{ github.event.inputs.auto-merge == 'false' && '' || '--auto-merge' }}"
+      - name: Run airbyte-ci connectors up-to-date [SCHEDULE]
+        if: github.event_name == 'schedule'
+        id: airbyte-ci-connectors-up-to-date-schedule
+        uses: ./.github/actions/run-airbyte-ci
+        with:
+          context: "master"
+          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_3 }}
+          docker_hub_password: ${{ secrets.DOCKER_HUB_PASSWORD }}
+          docker_hub_username: ${{ secrets.DOCKER_HUB_USERNAME }}
+          gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}
+          gcs_credentials: ${{ secrets.METADATA_SERVICE_PROD_GCS_CREDENTIALS }}
+          github_token: ${{ secrets.GH_PAT_MAINTENANCE_OSS }}
+          sentry_dsn: ${{ secrets.SENTRY_AIRBYTE_CI_DSN }}
+          s3_build_cache_access_key_id: ${{ secrets.SELF_RUNNER_AWS_ACCESS_KEY_ID }}
+          s3_build_cache_secret_key: ${{ secrets.SELF_RUNNER_AWS_SECRET_ACCESS_KEY }}
+          subcommand: 'connectors --concurrency=10 --language=python --language=low-code --support-level=community --support-level=certified --metadata-query="\"source-declarative-manifest\" not in data.dockerRepository"  --metadata-query="\"-rc.\" not in data.dockerImageTag" up-to-date --create-prs --auto-merge'

.github/workflows/connectors_version_increment_check.yml

@@ -23,7 +23,7 @@ jobs:
     name: Connectors Version Increment Check
     runs-on: connector-test-large
     if: github.event.pull_request.head.repo.fork != true
-    timeout-minutes: 12
+    timeout-minutes: 22
     steps:
       - name: Checkout Airbyte
         uses: actions/checkout@v4

.github/workflows/live_tests.yml

@@ -116,7 +116,7 @@ jobs:
         uses: ./.github/actions/run-airbyte-ci
         with:
           context: "manual"
-          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_4 }}
+          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_3 }}
           docker_hub_password: ${{ secrets.DOCKER_HUB_PASSWORD }}
           docker_hub_username: ${{ secrets.DOCKER_HUB_USERNAME }}
           gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}

.github/workflows/publish-bulk-cdk.yml

@@ -76,17 +76,6 @@ jobs:
           gradle-distribution-sha-256-sum-warning: false
           arguments: --scan :airbyte-cdk:bulk:bulkCdkBuild
 
-      - name: Integration test Bulk CDK
-        uses: burrunan/gradle-cache-action@v1
-        env:
-          CI: true
-        with:
-          read-only: true
-          job-id: bulk-cdk-publish
-          concurrent: true
-          gradle-distribution-sha-256-sum-warning: false
-          arguments: --scan :airbyte-cdk:bulk:bulkCdkIntegrationTest
-
       - name: Publish Poms and Jars to CloudRepo
         uses: burrunan/gradle-cache-action@v1
         env:

.github/workflows/publish_connectors.yml

@@ -30,7 +30,7 @@ jobs:
         uses: ./.github/actions/run-airbyte-ci
         with:
           context: "master"
-          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_3 }}
+          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_4 }}
           docker_hub_password: ${{ secrets.DOCKER_HUB_PASSWORD }}
           docker_hub_username: ${{ secrets.DOCKER_HUB_USERNAME }}
           gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}
@@ -53,7 +53,7 @@ jobs:
         uses: ./.github/actions/run-airbyte-ci
         with:
           context: "manual"
-          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_3 }}
+          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_4 }}
           docker_hub_password: ${{ secrets.DOCKER_HUB_PASSWORD }}
           docker_hub_username: ${{ secrets.DOCKER_HUB_USERNAME }}
           gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}

.github/workflows/regression_tests.yml

@@ -116,7 +116,7 @@ jobs:
         uses: ./.github/actions/run-airbyte-ci
         with:
           context: "manual"
-          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_4 }}
+          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_CACHE_3 }}
           docker_hub_password: ${{ secrets.DOCKER_HUB_PASSWORD }}
           docker_hub_username: ${{ secrets.DOCKER_HUB_USERNAME }}
           gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}

airbyte-cdk/bulk/build.gradle

@@ -61,10 +61,6 @@ allprojects {
     }
 }
 
-tasks.register('bulkCdkIntegrationTest').configure {
-    dependsOn allprojects.collect {it.tasks.matching { it.name == 'integrationTest' }}
-}
-
 if (buildNumberFile.exists()) {
     tasks.register('bulkCdkBuild').configure {
         dependsOn allprojects.collect {it.tasks.named('build')}

airbyte-cdk/bulk/core/base/build.gradle

@@ -50,3 +50,7 @@ dependencies {
     testFixturesApi 'io.micronaut.test:micronaut-test-junit5:4.5.0'
     testFixturesApi 'io.github.deblockt:json-diff:1.1.0'
 }
+
+test {
+    environment "PRESENT", "present-value"
+}

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

@@ -10,6 +10,6 @@ import io.airbyte.cdk.ssh.SshTunnelMethodConfiguration
 interface SshTunnelConfiguration {
     val realHost: String
     val realPort: Int
-    val sshTunnel: SshTunnelMethodConfiguration
+    val sshTunnel: SshTunnelMethodConfiguration?
     val sshConnectionOptions: SshConnectionOptions
 }

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

@@ -48,7 +48,7 @@ internal constructor(
 /** Creates an open [TunnelSession]. */
 fun createTunnelSession(
     remote: SshdSocketAddress,
-    sshTunnel: SshTunnelMethodConfiguration,
+    sshTunnel: SshTunnelMethodConfiguration?,
     connectionOptions: SshConnectionOptions,
 ): TunnelSession {
     if (sshTunnel is SshNoTunnelMethod) {
@@ -62,7 +62,8 @@ fun createTunnelSession(
         log.info { "Creating SSH client session." }
         val connectFuture: ConnectFuture =
             when (sshTunnel) {
-                SshNoTunnelMethod -> TODO("unreachable code")
+                SshNoTunnelMethod,
+                null -> TODO("unreachable code")
                 is SshKeyAuthTunnelMethod ->
                     client.connect(sshTunnel.user.trim(), sshTunnel.host.trim(), sshTunnel.port)
                 is SshPasswordAuthTunnelMethod ->

airbyte-cdk/bulk/core/base/src/main/resources/application.yaml

@@ -0,0 +1,4 @@
+airbyte:
+  file-transfer:
+    enabled: ${USE_FILE_TRANSFER:false}
+    staging-path: ${AIRBYTE_STAGING_DIRECTORY:/staging/files}

airbyte-cdk/bulk/core/base/src/test/kotlin/io/airbyte/cdk/initialization/TestApplicationYaml.kt

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.initialization
+
+import io.micronaut.context.annotation.Bean
+import io.micronaut.context.annotation.Factory
+import io.micronaut.context.annotation.Value
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest
+import jakarta.inject.Inject
+import kotlin.test.assertEquals
+import org.junit.jupiter.api.Test
+
+@MicronautTest
+class TestApplicationYaml {
+    @Inject lateinit var defaultValueBean: DefaultValueBean
+
+    @Test
+    fun testMainDefaultValue() {
+        assertEquals("/staging/files", defaultValueBean.stagingFolder)
+        assertEquals(false, defaultValueBean.fileTransferEnable)
+    }
+}
+
+data class DefaultValueBean(
+    val stagingFolder: String,
+    val fileTransferEnable: Boolean,
+)
+
+@Factory
+class TestFactory {
+    @Bean
+    fun defaultValueBean(
+        @Value("\${airbyte.file-transfer.staging-path}") stagingFolder: String,
+        @Value("\${airbyte.file-transfer.enabled}") fileTransferEnable: Boolean,
+    ): DefaultValueBean {
+        return DefaultValueBean(stagingFolder, fileTransferEnable)
+    }
+}

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

@@ -2,11 +2,11 @@
 package io.airbyte.cdk.discover
 
 import io.airbyte.cdk.data.AirbyteSchemaType
-import io.airbyte.cdk.data.IntCodec
 import io.airbyte.cdk.data.JsonDecoder
 import io.airbyte.cdk.data.JsonEncoder
 import io.airbyte.cdk.data.JsonStringCodec
 import io.airbyte.cdk.data.LeafAirbyteSchemaType
+import io.airbyte.cdk.data.LongCodec
 import io.airbyte.cdk.data.OffsetDateTimeCodec
 import java.time.OffsetDateTime
 
@@ -63,7 +63,6 @@ interface MetaField : FieldOrMetaField {
 enum class CommonMetaField(
     override val type: FieldType,
 ) : MetaField {
-    CDC_LSN(CdcStringMetaFieldType),
     CDC_UPDATED_AT(CdcOffsetDateTimeMetaFieldType),
     CDC_DELETED_AT(CdcOffsetDateTimeMetaFieldType),
     ;
@@ -80,8 +79,8 @@ data object CdcStringMetaFieldType : LosslessFieldType {
 
 data object CdcIntegerMetaFieldType : LosslessFieldType {
     override val airbyteSchemaType: AirbyteSchemaType = LeafAirbyteSchemaType.INTEGER
-    override val jsonEncoder: JsonEncoder<Int> = IntCodec
-    override val jsonDecoder: JsonDecoder<Int> = IntCodec
+    override val jsonEncoder: JsonEncoder<Long> = LongCodec
+    override val jsonDecoder: JsonDecoder<Long> = LongCodec
 }
 
 data object CdcOffsetDateTimeMetaFieldType : LosslessFieldType {

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

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.discover
+
+import com.fasterxml.jackson.databind.node.ObjectNode
+import io.airbyte.cdk.command.OpaqueStateValue
+import io.airbyte.cdk.read.Stream
+import io.airbyte.protocol.models.v0.AirbyteStream
+import java.time.OffsetDateTime
+
+/** [MetaField] schema definition and utilities, to be implemented by each source connector. */
+interface MetaFieldDecorator {
+
+    /** [MetaField] to use as a global cursor, if applicable. */
+    val globalCursor: MetaField?
+
+    /**
+     * All [MetaField]s to be found in [Global] stream records.
+     *
+     * This must include at least [globalCursor] if not null.
+     *
+     * Empty set when not applicable.
+     */
+    val globalMetaFields: Set<MetaField>
+
+    /** Convenience function for [AirbyteStreamFactory]. */
+    fun decorateAirbyteStream(airbyteStream: AirbyteStream) {
+        (airbyteStream.jsonSchema["properties"] as ObjectNode).apply {
+            for (metaField in globalMetaFields) {
+                set<ObjectNode>(metaField.id, metaField.type.airbyteSchemaType.asJsonSchema())
+            }
+        }
+        val globalCursorIdentifier: String = globalCursor?.id ?: return
+        airbyteStream.defaultCursorField = listOf(globalCursorIdentifier)
+        airbyteStream.sourceDefinedCursor = true
+    }
+
+    /**
+     * Modifies [recordData] by setting all [MetaField] values in global [Stream] feeds.
+     *
+     * This is required by the fact that records of a given stream may be emitted by both a [Stream]
+     * and a [Global] feed and the schemas must be the same. This implies that the records emitted
+     * by [Stream] must have [MetaField]s set to suitable values, even though that [Feed] has no
+     * awareness of the [Global] state.
+     *
+     * This method is called at most once per [Stream].
+     */
+    fun decorateRecordData(
+        /** Same value as emitted_at */
+        timestamp: OffsetDateTime,
+        /** Current state of the [Global] feed, if applicable. */
+        globalStateValue: OpaqueStateValue?,
+        stream: Stream,
+        recordData: ObjectNode
+    )
+}

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

@@ -27,6 +27,4 @@ interface MetadataQuerier : AutoCloseable {
         /** An implementation might open a connection to build a [MetadataQuerier] instance. */
         fun session(config: T): MetadataQuerier
     }
-
-    fun commonCursorOrNull(cursorColumnID: String): FieldOrMetaField?
 }

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

@@ -30,7 +30,7 @@ data class Global(
  */
 data class Stream(
     val id: StreamIdentifier,
-    val fields: List<Field>,
+    val schema: Set<FieldOrMetaField>,
     val configuredSyncMode: ConfiguredSyncMode,
     val configuredPrimaryKey: List<Field>?,
     val configuredCursor: FieldOrMetaField?,
@@ -43,6 +43,9 @@ data class Stream(
 
     override val label: String
         get() = id.toString()
+
+    val fields: List<Field>
+        get() = schema.filterIsInstance<Field>()
 }
 
 /** List of [Stream]s this [Feed] emits records for. */