Create public API thread pool (#12178)

airbytehq · Apr 19, 2024 · 1487103 · 1487103
1 parent 0340b27
commit 1487103
Show file tree

Hide file tree

Showing 9 changed files with 42 additions and 32 deletions.
diff --git a/...mmons-server/src/main/java/io/airbyte/commons/server/scheduling/AirbyteTaskExecutors.java b/...mmons-server/src/main/java/io/airbyte/commons/server/scheduling/AirbyteTaskExecutors.java
@@ -23,4 +23,10 @@ public interface AirbyteTaskExecutors extends TaskExecutors {
    */
   String SCHEDULER = "scheduler";
 
+  /**
+   * The name of the {@link java.util.concurrent.ExecutorService} used for endpoints that belong to
+   * the public API.
+   */
+  String PUBLIC_API = "public-api";
+
 }
diff --git a/...ver/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/ConnectionsController.kt b/...ver/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/ConnectionsController.kt
@@ -52,7 +52,7 @@ open class ConnectionsController(
   private val apiAuthorizationHelper: ApiAuthorizationHelper,
   private val currentUserService: CurrentUserService,
 ) : PublicConnectionsApi {
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicCreateConnection(connectionCreateRequest: ConnectionCreateRequest): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -177,7 +177,7 @@ open class ConnectionsController(
   }
 
   @Path("/{connectionId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicDeleteConnection(connectionId: UUID): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -210,7 +210,7 @@ open class ConnectionsController(
   }
 
   @Path("/{connectionId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicGetConnection(connectionId: UUID): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -237,7 +237,7 @@ open class ConnectionsController(
       .build()
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun listConnections(
     workspaceIds: List<UUID>?,
     includeDeleted: Boolean?,
@@ -275,7 +275,7 @@ open class ConnectionsController(
 
   @Patch
   @Path("/{connectionId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun patchConnection(
     @PathParam(value = "connectionId") connectionId: UUID,
     @Valid @Body @NotNull connectionPatchRequest:

diff --git a/...-server/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/DefaultController.kt b/...-server/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/DefaultController.kt
@@ -17,7 +17,7 @@ open class DefaultController() : PublicRootApi {
   @Value("\${airbyte.internal.documentation.host}")
   var documentationHost: String? = null
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun getDocumentation(): Response {
     return Response
       .status(302)

diff --git a/...er/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/DestinationsController.kt b/...er/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/DestinationsController.kt
@@ -45,7 +45,7 @@ open class DestinationsController(
   private val apiAuthorizationHelper: ApiAuthorizationHelper,
   private val currentUserService: CurrentUserService,
 ) : PublicDestinationsApi {
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicCreateDestination(destinationCreateRequest: DestinationCreateRequest): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -99,7 +99,7 @@ open class DestinationsController(
       .build()
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicDeleteDestination(destinationId: UUID): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -131,7 +131,7 @@ open class DestinationsController(
       .build()
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicGetDestination(destinationId: UUID): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -163,7 +163,7 @@ open class DestinationsController(
       .build()
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun listDestinations(
     workspaceIds: MutableList<UUID>?,
     includeDeleted: Boolean?,
@@ -201,7 +201,7 @@ open class DestinationsController(
 
   @Path("/{destinationId}")
   @Patch
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun patchDestination(
     destinationId: UUID,
     destinationPatchRequest: DestinationPatchRequest,
@@ -241,7 +241,7 @@ open class DestinationsController(
   }
 
   @Path("/{destinationId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun putDestination(
     destinationId: UUID,
     destinationPutRequest: DestinationPutRequest,

diff --git a/...yte-server/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/JobsController.kt b/...yte-server/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/JobsController.kt
@@ -48,7 +48,7 @@ open class JobsController(
 ) : PublicJobsApi {
   @DELETE
   @Path("/{jobId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicCancelJob(
     @PathParam("jobId") jobId: Long,
   ): Response {
@@ -83,7 +83,7 @@ open class JobsController(
       .build()
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicCreateJob(jobCreateRequest: JobCreateRequest): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -160,7 +160,7 @@ open class JobsController(
 
   @GET
   @Path("/{jobId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun getJob(
     @PathParam("jobId") jobId: Long,
   ): Response {
@@ -195,7 +195,7 @@ open class JobsController(
       .build()
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun listJobs(
     connectionId: UUID?,
     limit: Int?,

diff --git a/...-server/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/SourcesController.kt b/...-server/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/SourcesController.kt
@@ -47,7 +47,7 @@ open class SourcesController(
   private val apiAuthorizationHelper: ApiAuthorizationHelper,
   private val currentUserService: CurrentUserService,
 ) : PublicSourcesApi {
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicCreateSource(sourceCreateRequest: SourceCreateRequest): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -102,7 +102,7 @@ open class SourcesController(
       .build()
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicDeleteSource(sourceId: UUID): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -135,7 +135,7 @@ open class SourcesController(
       .build()
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicGetSource(sourceId: UUID): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -168,7 +168,7 @@ open class SourcesController(
       .build()
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun initiateOAuth(initiateOauthRequest: InitiateOauthRequest): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -180,7 +180,7 @@ open class SourcesController(
     return sourceService.controllerInitiateOAuth(initiateOauthRequest)
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun listSources(
     workspaceIds: MutableList<UUID>?,
     includeDeleted: Boolean?,
@@ -219,7 +219,7 @@ open class SourcesController(
 
   @Patch
   @Path("/{sourceId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun patchSource(
     sourceId: UUID,
     sourcePatchRequest: SourcePatchRequest,
@@ -259,7 +259,7 @@ open class SourcesController(
   }
 
   @Path("/{sourceId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun putSource(
     sourceId: UUID,
     sourcePutRequest: SourcePutRequest,

diff --git a/...-server/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/StreamsController.kt b/...-server/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/StreamsController.kt
@@ -47,7 +47,7 @@ class StreamsController(
     private val log: org.slf4j.Logger? = LoggerFactory.getLogger(StreamsController::class.java)
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun getStreamProperties(
     sourceId: UUID,
     destinationId: UUID?,

diff --git a/...rver/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/WorkspacesController.kt b/...rver/src/main/kotlin/io/airbyte/server/apis/publicapi/controllers/WorkspacesController.kt
@@ -37,7 +37,7 @@ open class WorkspacesController(
   private val currentUserService: CurrentUserService,
 ) : PublicWorkspacesApi {
   @Path("/{workspaceId}/oauthCredentials")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun createOrUpdateWorkspaceOAuthCredentials(
     workspaceId: UUID?,
     workspaceOAuthCredentialsRequest: WorkspaceOAuthCredentialsRequest?,
@@ -55,7 +55,7 @@ open class WorkspacesController(
     )
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicCreateWorkspace(workspaceCreateRequest: WorkspaceCreateRequest?): Response {
     // Now that we have orgs everywhere, ensure the user is at least an organization editor
     apiAuthorizationHelper.ensureUserHasAnyRequiredRoleOrThrow(
@@ -67,7 +67,7 @@ open class WorkspacesController(
   }
 
   @Path("/{workspaceId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicDeleteWorkspace(workspaceId: UUID?): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -80,7 +80,7 @@ open class WorkspacesController(
   }
 
   @Path("/{workspaceId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicGetWorkspace(workspaceId: UUID?): Response {
     val userId: UUID = currentUserService.currentUser.userId
     apiAuthorizationHelper.checkWorkspacePermissions(
@@ -92,7 +92,7 @@ open class WorkspacesController(
     return workspaceService.controllerGetWorkspace(workspaceId)
   }
 
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicListWorkspaces(
     workspaceIds: MutableList<UUID>?,
     includeDeleted: Boolean?,
@@ -117,7 +117,7 @@ open class WorkspacesController(
 
   @Patch
   @Path("/{workspaceId}")
-  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @ExecuteOn(AirbyteTaskExecutors.PUBLIC_API)
   override fun publicUpdateWorkspace(
     workspaceId: UUID?,
     workspaceUpdateRequest: WorkspaceUpdateRequest?,

diff --git a/airbyte-server/src/main/resources/application.yml b/airbyte-server/src/main/resources/application.yml
@@ -13,12 +13,16 @@ micronaut:
   env:
     cloud-deduction: true
   executors:
+    # We set our max pool size for the config DB to 20 in 10-values.yml files at the time of writing this.
     health:
       type: fixed
-      n-threads: ${HEALTH_TASK_EXECUTOR_THREADS:10} # Match the data source max pool size below
+      n-threads: ${HEALTH_TASK_EXECUTOR_THREADS:3}
     io:
       type: fixed
-      n-threads: ${IO_TASK_EXECUTOR_THREADS:10} # Match the data source max pool size below
+      n-threads: ${IO_TASK_EXECUTOR_THREADS:10}
+    public-api:
+      type: fixed
+      n-threads: ${PUBLIC_API_EXECUTOR_THREADS:5}
     scheduler:
       type: fixed
       n-threads: ${SCHEDULER_TASK_EXECUTOR_THREADS:25}