Bulk Core/Load CDK: Support for connector-type-specific non-config sp… (#45463)

Author: johnny-schmidt

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/load/src/main/kotlin/io/airbyte/cdk/spec/DestinationSpecification.kt

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.spec
+
+import io.airbyte.protocol.models.v0.ConnectorSpecification
+import io.airbyte.protocol.models.v0.DestinationSyncMode
+import io.micronaut.context.annotation.Replaces
+import io.micronaut.context.annotation.Requires
+import jakarta.inject.Singleton
+
+@Singleton
+@Replaces(IdentitySpecificationExtender::class)
+@Requires(env = ["destination"])
+class DestinationSpecificationExtender(private val spec: DestinationSpecification) :
+    SpecificationExtender {
+    override fun invoke(specification: ConnectorSpecification): ConnectorSpecification {
+        return specification
+            .withSupportedDestinationSyncModes(spec.supportedSyncModes)
+            .withSupportsIncremental(spec.supportsIncremental)
+    }
+}
+
+interface DestinationSpecification {
+    val supportedSyncModes: List<DestinationSyncMode>
+    val supportsIncremental: Boolean
+}