fix: command api and sync workflow usage of the api (#16489)

Author: benmoriceau

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

@@ -3934,7 +3934,14 @@ paths:
           content:
             application/json:
               schema:
+                title: GetWebhookConfigResponse
                 type: object
+                required:
+                  - value
+                properties:
+                  value:
+                    type: string
+
   /v1/jobs/job_failure:
     post:
       tags:
@@ -6364,7 +6371,7 @@ paths:
                   command_type:
                     type: string
                   command_input:
-                    type: object
+                    type: string
                   workspace_id:
                     type: string
                     format: uuid

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

@@ -342,7 +342,7 @@ private void saveAttemptSyncConfig(final long jobId, final int attemptNumber, fi
         .syncConfig(apiPojoConverters.attemptSyncConfigToApi(attemptSyncConfig, connectionId)));
   }
 
-  public Object getJobWebhookConfig(final long jobId) throws IOException {
+  public JobWebhookConfig getJobWebhookConfig(final long jobId) throws IOException {
     final Job job = jobPersistence.getJob(jobId);
     final JobConfig jobConfig = job.getConfig();
     if (jobConfig == null) {

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

@@ -92,7 +92,7 @@ class CommandApiController(
       id(commandGetRequest?.id)
       command?.let {
         commandType(it.commandType)
-        commandInput(it.commandInput)
+        commandInput(it.commandInput.toString())
         workspaceId(it.workspaceId)
         workloadId(it.workloadId)
         organizationId(it.organizationId)

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

@@ -11,6 +11,7 @@ import io.airbyte.api.model.generated.ConnectionIdRequestBody
 import io.airbyte.api.model.generated.ConnectionJobRequestBody
 import io.airbyte.api.model.generated.DeleteStreamResetRecordsForJobRequest
 import io.airbyte.api.model.generated.GetWebhookConfigRequest
+import io.airbyte.api.model.generated.GetWebhookConfigResponse
 import io.airbyte.api.model.generated.InternalOperationResult
 import io.airbyte.api.model.generated.JobCreate
 import io.airbyte.api.model.generated.JobDebugInfoRead
@@ -31,6 +32,7 @@ import io.airbyte.api.problems.throwable.generated.ApiNotImplementedInOssProblem
 import io.airbyte.commons.auth.AuthRoleConstants
 import io.airbyte.commons.auth.generated.Intent
 import io.airbyte.commons.auth.permissions.RequiresIntent
+import io.airbyte.commons.json.Jsons
 import io.airbyte.commons.server.handlers.JobHistoryHandler
 import io.airbyte.commons.server.handlers.JobInputHandler
 import io.airbyte.commons.server.handlers.JobsHandler
@@ -143,9 +145,11 @@ open class JobsApiController(
   @Post("/getWebhookConfig")
   @Secured(AuthRoleConstants.READER, AuthRoleConstants.WORKSPACE_READER, AuthRoleConstants.ORGANIZATION_READER)
   @ExecuteOn(AirbyteTaskExecutors.IO)
-  override fun getWebhookConfig(getWebhookConfigRequest: GetWebhookConfigRequest): Any? =
+  override fun getWebhookConfig(
+    @Body getWebhookConfigRequest: GetWebhookConfigRequest,
+  ): GetWebhookConfigResponse? =
     execute {
-      jobInputHandler.getJobWebhookConfig(getWebhookConfigRequest.jobId)
+      GetWebhookConfigResponse().value(Jsons.serialize(jobInputHandler.getJobWebhookConfig(getWebhookConfigRequest.jobId)))
     }
 
   @Post("/job_failure")

airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/activities/ConfigFetchActivityImpl.java

@@ -320,9 +320,9 @@ public Boolean isWorkspaceTombstone(UUID connectionId) {
   public GetWebhookConfigOutput getWebhookConfig(GetWebhookConfigInput input) {
     try {
       final JobWebhookConfig jobWebhookConfig = Jsons.deserialize(
-          airbyteApiClient.getJobsApi().getWebhookConfig(new GetWebhookConfigRequest(input.getJobId())).toString(), JobWebhookConfig.class);
+          airbyteApiClient.getJobsApi().getWebhookConfig(new GetWebhookConfigRequest(input.getJobId())).getValue(), JobWebhookConfig.class);
       return new GetWebhookConfigOutput(jobWebhookConfig.getOperationSequence(), jobWebhookConfig.getWebhookOperationConfigs());
-    } catch (IOException e) {
+    } catch (Exception e) {
       log.warn("Fail to get the webhook config.", e);
       throw new RuntimeException(e);
     }

airbyte-workers/src/test/kotlin/io/airbyte/workers/commands/ReplicationCommandTest.kt

@@ -99,9 +99,9 @@ class ReplicationCommandTest {
         commandType = "replicate",
         commandInput =
           Jsons
-            .jsonNode(
+            .serialize(
               ReplicationCommandApiInput.ReplicationApiInput(connectionId, jobId.toString(), attemptId, CatalogDiff()),
-            ).toString(),
+            ),
         workloadId = workloadId,
         organizationId = organizationId,
         createdAt = java.time.OffsetDateTime.now(),
@@ -161,9 +161,9 @@ class ReplicationCommandTest {
         commandType = "replicate",
         commandInput =
           Jsons
-            .jsonNode(
+            .serialize(
               ReplicationCommandApiInput.ReplicationApiInput(connectionId, jobId.toString(), attemptId, CatalogDiff()),
-            ).toString(),
+            ),
         workloadId = workloadId,
         organizationId = organizationId,
         createdAt = java.time.OffsetDateTime.now(),
@@ -232,9 +232,9 @@ class ReplicationCommandTest {
         commandType = "replicate",
         commandInput =
           Jsons
-            .jsonNode(
+            .serialize(
               ReplicationCommandApiInput.ReplicationApiInput(connectionId, jobId.toString(), attemptId, CatalogDiff()),
-            ).toString(),
+            ),
         workloadId = workloadId,
         organizationId = organizationId,
         createdAt = java.time.OffsetDateTime.now(),
@@ -275,9 +275,9 @@ class ReplicationCommandTest {
         commandType = "replicate",
         commandInput =
           Jsons
-            .jsonNode(
+            .serialize(
               ReplicationCommandApiInput.ReplicationApiInput(connectionId, jobId.toString(), attemptId, CatalogDiff()),
-            ).toString(),
+            ),
         workloadId = workloadId,
         organizationId = organizationId,
         createdAt = java.time.OffsetDateTime.now(),