feat(api): Idempotent Resource Creation

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

@@ -36,6 +36,7 @@ object ConnectionCreateMapper {
     connectionCreateOss.sourceId = connectionCreateRequest.sourceId
     connectionCreateOss.destinationId = connectionCreateRequest.destinationId
     connectionCreateOss.name = connectionCreateRequest.name
+    connectionCreateOss.idempotencyKey = connectionCreateRequest.idempotencyKey
     connectionCreateOss.nonBreakingChangesPreference =
       ConnectionHelper.convertNonBreakingSchemaUpdatesBehaviorEnum(
         connectionCreateRequest.nonBreakingSchemaUpdatesBehavior,

airbyte-api-server/src/main/kotlin/io/airbyte/api/server/services/DestinationService.kt

@@ -115,6 +115,7 @@ class DestinationServiceImpl(private val configApiClient: ConfigApiClient, priva
     destinationCreateOss.destinationDefinitionId = destinationDefinitionId
     destinationCreateOss.workspaceId = destinationCreateRequest.workspaceId
     destinationCreateOss.connectionConfiguration = destinationCreateRequest.configuration
+    destinationCreateOss.idempotencyKey = destinationCreateRequest.idempotencyKey
 
     val response =
       try {

airbyte-api-server/src/main/kotlin/io/airbyte/api/server/services/SourceService.kt

@@ -122,6 +122,7 @@ open class SourceServiceImpl(
     sourceCreateOss.workspaceId = sourceCreateRequest.workspaceId
     sourceCreateOss.connectionConfiguration = sourceCreateRequest.configuration
     sourceCreateOss.secretId = sourceCreateRequest.secretId
+    sourceCreateOss.idempotencyKey = sourceCreateRequest.idempotencyKey
 
     val response =
       try {

airbyte-api-server/src/main/kotlin/io/airbyte/api/server/services/WorkspaceService.kt

@@ -136,7 +136,7 @@ open class WorkspaceServiceImpl(
     authorization: String?,
     userInfo: String?,
   ): WorkspaceResponse {
-    val workspaceCreate = WorkspaceCreate().name(workspaceCreateRequest.name)
+    val workspaceCreate = WorkspaceCreate().name(workspaceCreateRequest.name).idempotencyKey(workspaceCreateRequest.idempotencyKey)
     val workspaceReadHttpResponse =
       try {
         configApiClient.createWorkspace(workspaceCreate, authorization, userInfo)

airbyte-api/src/main/openapi/api.yaml

@@ -1221,6 +1221,8 @@ components:
           $ref: "#/components/schemas/NonBreakingSchemaUpdatesBehaviorEnum"
         status:
           $ref: "#/components/schemas/ConnectionStatusEnum"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     ConnectionPatchRequest:
       type: object
       properties:
@@ -1299,6 +1301,8 @@ components:
         secretId:
           description: Optional secretID obtained through the public API OAuth redirect flow.
           type: string
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
       x-implements: io.airbyte.api.common.ConfigurableActor
     SourcePutRequest:
       required:
@@ -1586,6 +1590,8 @@ components:
           type: string
         configuration:
           $ref: "#/components/schemas/DestinationConfiguration"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
       x-implements: io.airbyte.api.common.ConfigurableActor
     DestinationPatchRequest:
       type: object
@@ -1614,6 +1620,8 @@ components:
         name:
           description: Name of the workspace
           type: string
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     WorkspaceUpdateRequest:
       required:
         - name
@@ -1811,3 +1819,8 @@ components:
       enum:
         - source
         - destination
+    IdempotencyKey:
+      type: string
+      format: uuid
+      description: |
+        An optional UUID that can be provided in a create request to prevent creating duplicates of the same resource

airbyte-api/src/main/openapi/config.yaml

@@ -4428,6 +4428,8 @@ components:
             $ref: "#/components/schemas/WebhookConfigWrite"
         organizationId:
           $ref: "#/components/schemas/OrganizationId"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     WorkspaceCreateWithId:
       type: object
       required:
@@ -4464,6 +4466,8 @@ components:
             $ref: "#/components/schemas/WebhookConfigWrite"
         organizationId:
           $ref: "#/components/schemas/OrganizationId"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     WebhookConfigWrite:
       type: object
       properties:
@@ -4997,6 +5001,8 @@ components:
           type: string
         resourceRequirements:
           $ref: "#/components/schemas/ActorDefinitionResourceRequirements"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     SourceDefinitionUpdate:
       type: object
       description: Update the SourceDefinition. Currently, the only allowed attribute to update is the default docker image version.
@@ -5090,6 +5096,8 @@ components:
           $ref: "#/components/schemas/ScopeType"
         sourceDefinition:
           $ref: "#/components/schemas/SourceDefinitionCreate"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     SourceDefinitionIdWithWorkspaceId:
       type: object
       required:
@@ -5280,6 +5288,8 @@ components:
         secretId:
           example: "airbyte_oauth_workspace_0509f049-d671-48cb-8105-0a23d47e6db6_secret_e0d38206-034e-4d75-9d21-da5a99b02826_v1"
           type: string
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     SourceDiscoverSchemaRequestBody:
       type: object
       required:
@@ -5478,6 +5488,8 @@ components:
           type: string
         resourceRequirements:
           $ref: "#/components/schemas/ActorDefinitionResourceRequirements"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     DestinationDefinitionUpdate:
       type: object
       required:
@@ -5558,6 +5570,8 @@ components:
           $ref: "#/components/schemas/ScopeId"
         scopeType:
           $ref: "#/components/schemas/ScopeType"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     DestinationDefinitionIdWithWorkspaceId:
       type: object
       required:
@@ -5657,6 +5671,8 @@ components:
           $ref: "#/components/schemas/DestinationDefinitionId"
         connectionConfiguration:
           $ref: "#/components/schemas/DestinationConfiguration"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     DestinationUpdate:
       type: object
       required:
@@ -5969,6 +5985,8 @@ components:
           type: boolean
         nonBreakingChangesPreference:
           $ref: "#/components/schemas/NonBreakingChangesPreference"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     WebBackendConnectionCreate:
       type: object
       required:
@@ -6020,6 +6038,8 @@ components:
           $ref: "#/components/schemas/Geography"
         nonBreakingChangesPreference:
           $ref: "#/components/schemas/NonBreakingChangesPreference"
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     ConnectionStateCreateOrUpdate:
       type: object
       required:
@@ -6687,6 +6707,8 @@ components:
           type: boolean
         orgLevelBilling:
           type: boolean
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     OrganizationRead:
       type: object
       required:
@@ -7038,7 +7060,8 @@ components:
         news:
           type: boolean
           default: false
-
+        idempotencyKey:
+          $ref: "#/components/schemas/IdempotencyKey"
     UserGetOrCreateByAuthIdResponse:
       type: object
       required:
@@ -9159,7 +9182,11 @@ components:
       properties:
         access_token:
           type: string
-
+    IdempotencyKey:
+      type: string
+      format: uuid
+      description: |
+        An optional UUID that can be provided in a create request to prevent creating duplicates of the same resource
   responses:
     NotFoundResponse:
       description: Object with given id was not found.

airbyte-bootloader/src/test/java/io/airbyte/bootloader/BootloaderTest.java

@@ -95,7 +95,7 @@ class BootloaderTest {
 
   // ⚠️ This line should change with every new migration to show that you meant to make a new
   // migration to the prod database
-  private static final String CURRENT_CONFIGS_MIGRATION_VERSION = "0.50.33.015";
+  private static final String CURRENT_CONFIGS_MIGRATION_VERSION = "0.50.33.016";
   private static final String CURRENT_JOBS_MIGRATION_VERSION = "0.50.4.001";
   private static final String CDK_VERSION = "1.2.3";
 

airbyte-commons-server/src/main/java/io/airbyte/commons/server/handlers/ConnectionsHandler.java

@@ -87,6 +87,7 @@
 import io.airbyte.config.persistence.ActorDefinitionVersionHelper;
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.data.services.ConnectionService;
 import io.airbyte.featureflag.CheckWithCatalog;
 import io.airbyte.featureflag.FeatureFlagClient;
 import io.airbyte.featureflag.Workspace;
@@ -151,6 +152,7 @@ public class ConnectionsHandler {
   private final ActorDefinitionVersionHelper actorDefinitionVersionHelper;
   private final ConnectorDefinitionSpecificationHandler connectorSpecHandler;
   private final JobNotifier jobNotifier;
+  private final ConnectionService connectionService;
   private final Integer maxDaysOfOnlyFailedJobsBeforeConnectionDisable;
   private final Integer maxFailedJobsInARowBeforeConnectionDisable;
   private final int maxJobLookback = 10;
@@ -168,6 +170,7 @@ public ConnectionsHandler(
                             final ActorDefinitionVersionHelper actorDefinitionVersionHelper,
                             final ConnectorDefinitionSpecificationHandler connectorSpecHandler,
                             final JobNotifier jobNotifier,
+                            final ConnectionService connectionService,
                             @Value("${airbyte.server.connection.disable.max-days}") final Integer maxDaysOfOnlyFailedJobsBeforeConnectionDisable,
                             @Value("${airbyte.server.connection.disable.max-jobs}") final Integer maxFailedJobsInARowBeforeConnectionDisable) {
     this.jobPersistence = jobPersistence;
@@ -181,6 +184,7 @@ public ConnectionsHandler(
     this.actorDefinitionVersionHelper = actorDefinitionVersionHelper;
     this.connectorSpecHandler = connectorSpecHandler;
     this.jobNotifier = jobNotifier;
+    this.connectionService = connectionService;
     this.maxDaysOfOnlyFailedJobsBeforeConnectionDisable = maxDaysOfOnlyFailedJobsBeforeConnectionDisable;
     this.maxFailedJobsInARowBeforeConnectionDisable = maxFailedJobsInARowBeforeConnectionDisable;
   }
@@ -432,7 +436,13 @@ private boolean warningPreviouslySentForMaxDays(final int numFailures,
 
   public ConnectionRead createConnection(final ConnectionCreate connectionCreate)
       throws JsonValidationException, IOException, ConfigNotFoundException {
-
+    final UUID idempotencyKey = connectionCreate.getIdempotencyKey();
+    if (idempotencyKey != null) {
+      final Optional<StandardSync> found = connectionService.getStandardSyncByIdempotencyKey(idempotencyKey);
+      if (found.isPresent()) {
+        return buildConnectionRead(found.get().getConnectionId());
+      }
+    }
     // Validate source and destination
     final SourceConnection sourceConnection = configRepository.getSourceConnection(connectionCreate.getSourceId());
     final DestinationConnection destinationConnection = configRepository.getDestinationConnection(connectionCreate.getDestinationId());
@@ -469,6 +479,7 @@ public ConnectionRead createConnection(final ConnectionCreate connectionCreate)
         .withSourceCatalogId(connectionCreate.getSourceCatalogId())
         .withGeography(getGeographyFromConnectionCreateOrWorkspace(connectionCreate))
         .withBreakingChange(false)
+        .withIdempotencyKey(idempotencyKey)
         .withNonBreakingChangesPreference(
             ApiPojoConverters.toPersistenceNonBreakingChangesPreference(connectionCreate.getNonBreakingChangesPreference()));
     if (connectionCreate.getResourceRequirements() != null) {

airbyte-commons-server/src/main/java/io/airbyte/commons/server/handlers/DestinationDefinitionsHandler.java

@@ -37,6 +37,7 @@
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.config.specs.RemoteDefinitionsProvider;
+import io.airbyte.data.services.DestinationService;
 import io.airbyte.featureflag.DestinationDefinition;
 import io.airbyte.featureflag.FeatureFlagClient;
 import io.airbyte.featureflag.HideActorDefinitionFromList;
@@ -54,6 +55,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -68,6 +70,7 @@
 public class DestinationDefinitionsHandler {
 
   private final ConfigRepository configRepository;
+  private final DestinationService destinationService;
   private final Supplier<UUID> uuidSupplier;
   private final ActorDefinitionHandlerHelper actorDefinitionHandlerHelper;
   private final RemoteDefinitionsProvider remoteDefinitionsProvider;
@@ -82,14 +85,16 @@ public DestinationDefinitionsHandler(final ConfigRepository configRepository,
                                        final RemoteDefinitionsProvider remoteDefinitionsProvider,
                                        final DestinationHandler destinationHandler,
                                        final SupportStateUpdater supportStateUpdater,
-                                       final FeatureFlagClient featureFlagClient) {
+                                       final FeatureFlagClient featureFlagClient,
+                                       final DestinationService destinationService) {
     this.configRepository = configRepository;
     this.uuidSupplier = uuidSupplier;
     this.actorDefinitionHandlerHelper = actorDefinitionHandlerHelper;
     this.remoteDefinitionsProvider = remoteDefinitionsProvider;
     this.destinationHandler = destinationHandler;
     this.supportStateUpdater = supportStateUpdater;
     this.featureFlagClient = featureFlagClient;
+    this.destinationService = destinationService;
   }
 
   @VisibleForTesting
@@ -233,6 +238,20 @@ public DestinationDefinitionRead getDestinationDefinitionForScope(final ActorDef
 
   public DestinationDefinitionRead createCustomDestinationDefinition(final CustomDestinationDefinitionCreate customDestinationDefinitionCreate)
       throws IOException {
+    final UUID idempotencyKey = customDestinationDefinitionCreate.getIdempotencyKey();
+    if (idempotencyKey != null) {
+      final Optional<StandardDestinationDefinition> found = destinationService.getStandardDestinationDefinitionByIdempotencyKey(idempotencyKey);
+      if (found.isPresent()) {
+        final StandardDestinationDefinition destinationDefinition = found.get();
+        final ActorDefinitionVersion actorDefinitionVersion;
+        try {
+          actorDefinitionVersion = configRepository.getActorDefinitionVersion(destinationDefinition.getDefaultVersionId());
+          return buildDestinationDefinitionRead(destinationDefinition, actorDefinitionVersion);
+        } catch (ConfigNotFoundException e) {
+          // not found, so must create
+        }
+      }
+    }
     final UUID id = uuidSupplier.get();
     final DestinationDefinitionCreate destinationDefCreate = customDestinationDefinitionCreate.getDestinationDefinition();
     final ActorDefinitionVersion actorDefinitionVersion =