destination-e2e: merge cloud and OSS

airbytehq · Sep 18, 2024 · 236bd09 · 236bd09
1 parent 25bde21
commit 236bd09
Show file tree

Hide file tree

Showing 16 changed files with 74 additions and 188 deletions.
diff --git a/...te-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/BaseConnector.kt b/...te-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/BaseConnector.kt
@@ -4,11 +4,19 @@
 package io.airbyte.cdk.integrations
 
 import io.airbyte.cdk.integrations.base.Integration
+import io.airbyte.cdk.integrations.base.adaptive.AdaptiveSourceRunner
+import io.airbyte.commons.features.EnvVariableFeatureFlags
+import io.airbyte.commons.features.FeatureFlags
 import io.airbyte.commons.json.Jsons
 import io.airbyte.commons.resources.MoreResources
 import io.airbyte.protocol.models.v0.ConnectorSpecification
 
 abstract class BaseConnector : Integration {
+    open val featureFlags: FeatureFlags = EnvVariableFeatureFlags()
+
+    val isCloudDeployment
+        get() =
+            AdaptiveSourceRunner.CLOUD_MODE.equals(featureFlags.deploymentMode(), ignoreCase = true)
     /**
      * By convention the spec is stored as a resource for java connectors. That resource is called
      * spec.json.

diff --git a/.../kotlin/io/airbyte/cdk/integrations/standardtest/destination/DestinationAcceptanceTest.kt b/.../kotlin/io/airbyte/cdk/integrations/standardtest/destination/DestinationAcceptanceTest.kt
@@ -17,6 +17,8 @@ import io.airbyte.cdk.integrations.standardtest.destination.argproviders.util.Ar
 import io.airbyte.cdk.integrations.standardtest.destination.comparator.BasicTestDataComparator
 import io.airbyte.cdk.integrations.standardtest.destination.comparator.TestDataComparator
 import io.airbyte.commons.features.EnvVariableFeatureFlags
+import io.airbyte.commons.features.FeatureFlags
+import io.airbyte.commons.features.FeatureFlagsWrapper
 import io.airbyte.commons.jackson.MoreMappers
 import io.airbyte.commons.json.Jsons
 import io.airbyte.commons.lang.Exceptions
@@ -98,6 +100,13 @@ abstract class DestinationAcceptanceTest(
     protected var testSchemas: HashSet<String> = HashSet()
 
     private lateinit var testEnv: TestDestinationEnv
+    protected open val isCloudTest: Boolean = true
+    protected val featureFlags: FeatureFlags =
+        if (isCloudTest) {
+            FeatureFlagsWrapper.overridingDeploymentMode(EnvVariableFeatureFlags(), "CLOUD")
+        } else {
+            FeatureFlagsWrapper.overridingDeploymentMode(EnvVariableFeatureFlags(), "OSS")
+        }
 
     private lateinit var jobRoot: Path
     private lateinit var processFactory: ProcessFactory
@@ -1911,7 +1920,7 @@ abstract class DestinationAcceptanceTest(
                         null,
                         null,
                         false,
-                        EnvVariableFeatureFlags()
+                        featureFlags
                     )
                 )
                 .run(JobGetSpecConfig().withDockerImage(imageName), jobRoot)
@@ -1931,7 +1940,7 @@ abstract class DestinationAcceptanceTest(
                     null,
                     null,
                     false,
-                    EnvVariableFeatureFlags()
+                    featureFlags
                 ),
                 mConnectorConfigUpdater
             )
@@ -1953,7 +1962,7 @@ abstract class DestinationAcceptanceTest(
                             null,
                             null,
                             false,
-                            EnvVariableFeatureFlags()
+                            featureFlags
                         ),
                         mConnectorConfigUpdater
                     )
@@ -1981,7 +1990,7 @@ abstract class DestinationAcceptanceTest(
                         null,
                         null,
                         false,
-                        EnvVariableFeatureFlags()
+                        featureFlags
                     )
             )
         }
@@ -1997,7 +2006,7 @@ abstract class DestinationAcceptanceTest(
                     null,
                     null,
                     false,
-                    EnvVariableFeatureFlags()
+                    featureFlags
                 )
         )
     }

diff --git a/...urces/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/AbstractDbSource.kt b/...urces/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/AbstractDbSource.kt
@@ -22,8 +22,6 @@ import io.airbyte.cdk.integrations.util.ApmTraceUtils.addExceptionToTrace
 import io.airbyte.cdk.integrations.util.ConnectorExceptionUtil
 import io.airbyte.commons.exceptions.ConfigErrorException
 import io.airbyte.commons.exceptions.ConnectionErrorException
-import io.airbyte.commons.features.EnvVariableFeatureFlags
-import io.airbyte.commons.features.FeatureFlags
 import io.airbyte.commons.functional.CheckedConsumer
 import io.airbyte.commons.lang.Exceptions
 import io.airbyte.commons.stream.AirbyteStreamUtils
@@ -49,8 +47,6 @@ private val LOGGER = KotlinLogging.logger {}
 abstract class AbstractDbSource<DataType, Database : AbstractDatabase?>
 protected constructor(driverClassName: String) :
     JdbcConnector(driverClassName), Source, AutoCloseable {
-    // TODO: Remove when the flag is not use anymore
-    var featureFlags: FeatureFlags = EnvVariableFeatureFlags()
 
     @Trace(operationName = CHECK_TRACE_OPERATION_NAME)
     @Throws(Exception::class)

diff --git a/airbyte-integrations/connectors/destination-dev-null/README.md b/airbyte-integrations/connectors/destination-dev-null/README.md
diff --git a/airbyte-integrations/connectors/destination-dev-null/build.gradle b/airbyte-integrations/connectors/destination-dev-null/build.gradle
diff --git a/airbyte-integrations/connectors/destination-dev-null/icon.svg b/airbyte-integrations/connectors/destination-dev-null/icon.svg
diff --git a/airbyte-integrations/connectors/destination-dev-null/metadata.yaml b/airbyte-integrations/connectors/destination-dev-null/metadata.yaml
diff --git a/...v-null/src/main/kotlin/io/airbyte/integrations/destination/dev_null/DevNullDestination.kt b/...v-null/src/main/kotlin/io/airbyte/integrations/destination/dev_null/DevNullDestination.kt
diff --git a/airbyte-integrations/connectors/destination-e2e-test/metadata.yaml b/airbyte-integrations/connectors/destination-e2e-test/metadata.yaml
@@ -2,7 +2,7 @@ data:
   connectorSubtype: unknown
   connectorType: destination
   definitionId: 2eb65e87-983a-4fd7-b3e3-9d9dc6eb8537
-  dockerImageTag: 0.5.0
+  dockerImageTag: 0.6.0
   dockerRepository: airbyte/destination-e2e-test
   githubIssueLabel: destination-e2e-test
   icon: airbyte.svg

diff --git a/...-test/src/main/kotlin/io/airbyte/integrations/destination/e2e_test/TestingDestinations.kt b/...-test/src/main/kotlin/io/airbyte/integrations/destination/e2e_test/TestingDestinations.kt
@@ -4,11 +4,15 @@
 package io.airbyte.integrations.destination.e2e_test
 
 import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.node.ArrayNode
+import com.fasterxml.jackson.databind.node.ObjectNode
 import com.google.common.collect.ImmutableMap
 import io.airbyte.cdk.integrations.BaseConnector
 import io.airbyte.cdk.integrations.base.AirbyteMessageConsumer
 import io.airbyte.cdk.integrations.base.Destination
 import io.airbyte.cdk.integrations.base.IntegrationRunner
+import io.airbyte.commons.features.EnvVariableFeatureFlags
+import io.airbyte.commons.features.FeatureFlags
 import io.airbyte.protocol.models.v0.AirbyteConnectionStatus
 import io.airbyte.protocol.models.v0.AirbyteMessage
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog
@@ -19,13 +23,14 @@ import org.slf4j.LoggerFactory
 class TestingDestinations
 @JvmOverloads
 constructor(
+    override val featureFlags: FeatureFlags = EnvVariableFeatureFlags(),
     private val destinationMap: Map<TestDestinationType, Destination> =
         ImmutableMap.builder<TestDestinationType, Destination>()
             .put(TestDestinationType.LOGGING, LoggingDestination())
             .put(TestDestinationType.THROTTLED, ThrottledDestination())
             .put(TestDestinationType.SILENT, SilentDestination())
             .put(TestDestinationType.FAILING, FailAfterNDestination())
-            .build()
+            .build(),
 ) : BaseConnector(), Destination {
     enum class TestDestinationType {
         LOGGING,
@@ -42,6 +47,33 @@ constructor(
         ]
     }
 
+    override fun spec(): ConnectorSpecification {
+        if (isCloudDeployment) {
+            return super.spec()
+        } else {
+            /** 1. Update the title. 2. Only keep the "silent" mode. */
+            val spec = super.spec()
+
+            (spec.connectionSpecification as ObjectNode).put("title", DEV_NULL_DESTINATION_TITLE)
+
+            val properties =
+                spec.connectionSpecification["properties"]["test_destination"] as ObjectNode
+            val types = properties["oneOf"] as ArrayNode
+            val typesIterator = types.elements()
+            while (typesIterator.hasNext()) {
+                val typeNode = typesIterator.next()
+                if (
+                    !typeNode["properties"]["test_destination_type"]["const"]
+                        .asText()
+                        .equals("silent", ignoreCase = true)
+                ) {
+                    typesIterator.remove()
+                }
+            }
+            return spec
+        }
+    }
+
     @Throws(Exception::class)
     override fun getConsumer(
         config: JsonNode,
@@ -64,6 +96,7 @@ constructor(
 
     companion object {
         private val LOGGER: Logger = LoggerFactory.getLogger(TestingDestinations::class.java)
+        private const val DEV_NULL_DESTINATION_TITLE = "E2E Test (/dev/null) Destination Spec"
 
         @Throws(Exception::class)
         @JvmStatic

diff --git a/...test/src/main/kotlin/io/airbyte/integrations/destination/e2e_test/ThrottledDestination.kt b/...test/src/main/kotlin/io/airbyte/integrations/destination/e2e_test/ThrottledDestination.kt
@@ -18,7 +18,7 @@ import org.slf4j.LoggerFactory
  * This destination logs each record it receives. It sleeps for millis_per_record between accepting
  * each record. Useful for simulating backpressure / slow destination writes.
  */
-class ThrottledDestination : BaseConnector(), Destination {
+class ThrottledDestination() : BaseConnector(), Destination {
     override fun check(config: JsonNode): AirbyteConnectionStatus {
         return AirbyteConnectionStatus().withStatus(AirbyteConnectionStatus.Status.SUCCEEDED)
     }

diff --git a/..._null/DevNullDestinationAcceptanceTest.kt → ..._test/DevNullDestinationAcceptanceTest.kt b/..._null/DevNullDestinationAcceptanceTest.kt → ..._test/DevNullDestinationAcceptanceTest.kt
@@ -1,10 +1,9 @@
 /*
  * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
  */
-package io.airbyte.integrations.destination.dev_null
+package io.airbyte.integrations.destination.e2e_test
 
 import com.fasterxml.jackson.databind.JsonNode
-import io.airbyte.cdk.integrations.standardtest.destination.DestinationAcceptanceTest
 import io.airbyte.commons.json.Jsons
 import io.airbyte.protocol.models.v0.AirbyteMessage
 import io.airbyte.protocol.models.v0.AirbyteRecordMessage
@@ -13,8 +12,8 @@ import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 
-class DevNullDestinationAcceptanceTest : DestinationAcceptanceTest() {
-    override val imageName = "airbyte/destination-dev-null:dev"
+class DevNullDestinationAcceptanceTest : TestingSilentDestinationAcceptanceTest() {
+    override val isCloudTest = false
 
     override fun getConfig(): JsonNode {
         return Jsons.jsonNode(