fix: set destinationCatalogId when creating DA connection via public-api (#17656)

Author: pedroslopez

airbyte-commons-server/src/main/kotlin/io/airbyte/commons/server/services/DestinationDiscoverService.kt

@@ -88,6 +88,16 @@ class DestinationDiscoverService(
     return actorCatalogToDestinationCatalog(actorCatalog)
   }
 
+  fun getDestinationCatalogIfSupported(
+    destinationId: ActorId,
+    skipCache: Boolean = false,
+  ): DestinationCatalogWithId? =
+    try {
+      getDestinationCatalog(destinationId, skipCache)
+    } catch (_: DestinationDiscoverNotSupportedProblem) {
+      null
+    }
+
   /**
    * Retrieves the destination catalog associated with a given connection.
    *

airbyte-commons-server/src/test/kotlin/io/airbyte/commons/server/services/DestinationDiscoverServiceTest.kt

@@ -222,6 +222,33 @@ class DestinationDiscoverServiceTest {
     }
   }
 
+  @Nested
+  inner class GetDestinationCatalogIfSupported {
+    @Test
+    fun `should return null when destination version does not support data activation`() {
+      val destination =
+        DestinationConnection()
+          .withWorkspaceId(workspaceId)
+          .withDestinationDefinitionId(destinationDefinitionId)
+          .withDestinationId(destinationId.value)
+          .withConfiguration(Jsons.emptyObject())
+
+      val destinationDefinition = StandardDestinationDefinition()
+      val destinationVersion =
+        ActorDefinitionVersion()
+          .withDockerImageTag("1.0.1")
+          .withSupportsDataActivation(false)
+
+      every { destinationService.getDestinationConnection(destinationId.value) } returns destination
+      every { destinationService.getStandardDestinationDefinition(destinationDefinitionId) } returns destinationDefinition
+      every { actorDefinitionVersionHelper.getDestinationVersion(destinationDefinition, workspaceId, destinationId.value) } returns destinationVersion
+
+      val result = service.getDestinationCatalogIfSupported(destinationId)
+
+      result shouldBe null
+    }
+  }
+
   @Nested
   inner class WriteCatalogResult {
     @Test

airbyte-server/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/ConnectionsController.kt

@@ -11,8 +11,11 @@ import io.airbyte.api.model.generated.SourceDiscoverSchemaRead
 import io.airbyte.commons.auth.roles.AuthRoleConstants
 import io.airbyte.commons.server.authorization.RoleResolver
 import io.airbyte.commons.server.scheduling.AirbyteTaskExecutors
+import io.airbyte.commons.server.services.DestinationDiscoverService
 import io.airbyte.commons.server.support.AuthenticationId
 import io.airbyte.commons.server.support.CurrentUserService
+import io.airbyte.config.DestinationCatalog
+import io.airbyte.domain.models.ActorId
 import io.airbyte.publicApi.server.generated.apis.PublicConnectionsApi
 import io.airbyte.publicApi.server.generated.models.ConnectionCreateRequest
 import io.airbyte.publicApi.server.generated.models.ConnectionPatchRequest
@@ -48,6 +51,7 @@ open class ConnectionsController(
   private val connectionService: ConnectionService,
   private val sourceService: SourceService,
   private val destinationService: DestinationService,
+  private val destinationDiscoverService: DestinationDiscoverService,
   private val trackingHelper: TrackingHelper,
   private val roleResolver: RoleResolver,
   private val currentUserService: CurrentUserService,
@@ -105,6 +109,14 @@ open class ConnectionsController(
         userId,
       ) as List<DestinationSyncMode>
 
+    // get destination_catalog_id
+    val destinationCatalogId =
+      destinationDiscoverService
+        .getDestinationCatalogIfSupported(
+          destinationId = ActorId(connectionCreateRequest.destinationId),
+          skipCache = false,
+        )?.catalogId
+
     // refer to documentation to understand what we need to do for the catalog
     // https://docs.airbyte.com/understanding-airbyte/airbyte-protocol/#catalog
     var configuredCatalog = AirbyteCatalog()
@@ -135,6 +147,7 @@ open class ConnectionsController(
         connectionService.createConnection(
           validConnectionCreateRequest,
           catalogId!!,
+          destinationCatalogId?.value,
           finalConfiguredCatalog,
           destinationRead.workspaceId,
         )

airbyte-server/src/main/kotlin/io/airbyte/server/apis/publicapi/mappers/ConnectionCreateMapper.kt

@@ -29,6 +29,7 @@ object ConnectionCreateMapper {
   fun from(
     connectionCreateRequest: ConnectionCreateRequest,
     catalogId: UUID?,
+    destinationCatalogId: UUID?,
     configuredCatalog: AirbyteCatalog?,
   ): ConnectionCreate {
     val connectionCreateOss = ConnectionCreate()
@@ -67,6 +68,9 @@ object ConnectionCreateMapper {
     if (configuredCatalog != null) {
       connectionCreateOss.syncCatalog = configuredCatalog
     }
+    if (destinationCatalogId != null) {
+      connectionCreateOss.destinationCatalogId = destinationCatalogId
+    }
     if (connectionCreateRequest.status != null) {
       connectionCreateOss.status = ConnectionStatus.fromValue(connectionCreateRequest.status.toString())
     } else {

airbyte-server/src/main/kotlin/io/airbyte/server/apis/publicapi/services/ConnectionService.kt

@@ -35,6 +35,7 @@ interface ConnectionService {
   fun createConnection(
     connectionCreateRequest: ConnectionCreateRequest,
     catalogId: UUID,
+    destinationCatalogId: UUID?,
     configuredCatalog: AirbyteCatalog,
     workspaceId: UUID,
   ): ConnectionResponse
@@ -79,13 +80,15 @@ class ConnectionServiceImpl(
   override fun createConnection(
     connectionCreateRequest: ConnectionCreateRequest,
     catalogId: UUID,
+    destinationCatalogId: UUID?,
     configuredCatalog: AirbyteCatalog,
     workspaceId: UUID,
   ): ConnectionResponse {
     val connectionCreateOss: ConnectionCreate =
       ConnectionCreateMapper.from(
         connectionCreateRequest,
         catalogId,
+        destinationCatalogId,
         configuredCatalog,
       )
 

airbyte-server/src/test/kotlin/io/airbyte/server/apis/publicapi/controllers/ConnectionsControllerTest.kt

@@ -9,6 +9,7 @@ import io.airbyte.api.model.generated.AirbyteStreamAndConfiguration
 import io.airbyte.api.model.generated.DestinationRead
 import io.airbyte.api.model.generated.SourceDiscoverSchemaRead
 import io.airbyte.commons.server.authorization.RoleResolver
+import io.airbyte.commons.server.services.DestinationDiscoverService
 import io.airbyte.commons.server.support.CurrentUserService
 import io.airbyte.config.AuthenticatedUser
 import io.airbyte.publicApi.server.generated.models.ConnectionPatchRequest
@@ -37,6 +38,7 @@ class ConnectionsControllerTest {
   private val trackingHelper: TrackingHelper = mockk(relaxed = true)
   private val roleResolver: RoleResolver = mockk(relaxed = true)
   private val currentUserService: CurrentUserService = mockk()
+  private val destinationDiscoverService: DestinationDiscoverService = mockk(relaxed = true)
 
   @BeforeEach
   fun setUp() {
@@ -56,6 +58,7 @@ class ConnectionsControllerTest {
         trackingHelper = trackingHelper,
         roleResolver = roleResolver,
         currentUserService = currentUserService,
+        destinationDiscoverService = destinationDiscoverService,
       )
   }
 

airbyte-server/src/test/kotlin/io/airbyte/server/apis/publicapi/mappers/ConnectionCreateMapperTest.kt

@@ -24,6 +24,7 @@ class ConnectionCreateMapperTest {
   @Test
   fun testConnectionCreateMapper() {
     val catalogId = UUID.randomUUID()
+    val destinationCatalogId = UUID.randomUUID()
 
     val catalog =
       AirbyteCatalog().apply {
@@ -59,6 +60,7 @@ class ConnectionCreateMapperTest {
         this.prefix = "test"
         this.scheduleType = ConnectionScheduleType.CRON
         this.sourceCatalogId = catalogId
+        this.destinationCatalogId = destinationCatalogId
         this.syncCatalog = catalog
         this.status = ConnectionStatus.INACTIVE
         val connectionScheduleDataCron =
@@ -72,6 +74,6 @@ class ConnectionCreateMapperTest {
           }
         this.scheduleData = connectionScheduleData
       }
-    assertEquals(expectedOssConnectionCreateRequest, ConnectionCreateMapper.from(connectionCreateRequest, catalogId, catalog))
+    assertEquals(expectedOssConnectionCreateRequest, ConnectionCreateMapper.from(connectionCreateRequest, catalogId, destinationCatalogId, catalog))
   }
 }