chore: fix ui_commands queue config (#14865)

airbyte-commons-temporal/src/main/kotlin/io/airbyte/commons/temporal/TemporalClient.kt

@@ -6,6 +6,7 @@ package io.airbyte.commons.temporal
 import com.google.common.annotations.VisibleForTesting
 import com.google.protobuf.ByteString
 import io.airbyte.commons.annotation.InternalForTesting
+import io.airbyte.commons.temporal.config.TemporalQueueConfiguration
 import io.airbyte.commons.temporal.exception.DeletedWorkflowException
 import io.airbyte.commons.temporal.exception.UnreachableWorkflowException
 import io.airbyte.commons.temporal.scheduling.CheckCommandInput
@@ -40,7 +41,6 @@ import io.airbyte.metrics.lib.OssMetricsRegistry
 import io.airbyte.persistence.job.models.IntegrationLauncherConfig
 import io.airbyte.persistence.job.models.JobRunConfig
 import io.github.oshai.kotlinlogging.KotlinLogging
-import io.micronaut.context.annotation.Property
 import io.temporal.api.common.v1.WorkflowType
 import io.temporal.api.enums.v1.WorkflowExecutionStatus
 import io.temporal.api.workflowservice.v1.ListClosedWorkflowExecutionsRequest
@@ -87,7 +87,7 @@ data class ManualOperationResult(
 @Singleton
 class TemporalClient(
   @param:Named("workspaceRootTemporal") private val workspaceRoot: Path,
-  @param:Property(name = "airbyte.temporal.queues.ui-commands") private val uiCommandsQueue: String?,
+  private val queueConfiguration: TemporalQueueConfiguration,
   private val workflowClientWrapped: WorkflowClientWrapped,
   private val serviceStubsWrapped: WorkflowServiceStubsWrapped,
   private val streamResetPersistence: StreamResetPersistence,
@@ -581,7 +581,7 @@ class TemporalClient(
     val workflowOptions =
       WorkflowOptions
         .newBuilder()
-        .setTaskQueue(uiCommandsQueue)
+        .setTaskQueue(queueConfiguration.uiCommandsQueue)
         .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build())
         .setWorkflowId(String.format("%s_%s", input.type, jobRunConfig.getJobId()))
         .build()

airbyte-commons-temporal/src/main/kotlin/io/airbyte/commons/temporal/TemporalTaskQueueUtils.kt

@@ -8,6 +8,7 @@ package io.airbyte.commons.temporal
  * Singleton helper object that provides translation between a Temporal job type and
  * the Temporal queue name.
  */
+@Deprecated("Check and Discover queues are obsolete, SYNC should be migrated to TemporalQueueConfiguration")
 object TemporalTaskQueueUtils {
   val DEFAULT_SYNC_TASK_QUEUE = TemporalJobType.SYNC.name
   val DEFAULT_CHECK_TASK_QUEUE = TemporalJobType.CHECK_CONNECTION.name

airbyte-commons-temporal/src/main/kotlin/io/airbyte/commons/temporal/config/TemporalQueueConfiguration.kt

@@ -0,0 +1,8 @@
+package io.airbyte.commons.temporal.config
+
+import jakarta.inject.Singleton
+
+@Singleton
+class TemporalQueueConfiguration {
+  val uiCommandsQueue = "ui_commands"
+}

airbyte-commons-temporal/src/test/java/io/airbyte/commons/temporal/TemporalClientTest.java

@@ -26,6 +26,7 @@
 
 import com.google.common.collect.Sets;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.temporal.config.TemporalQueueConfiguration;
 import io.airbyte.commons.temporal.exception.DeletedWorkflowException;
 import io.airbyte.commons.temporal.scheduling.CheckCommandInput;
 import io.airbyte.commons.temporal.scheduling.ConnectionManagerWorkflow;
@@ -90,7 +91,7 @@ public class TemporalClientTest {
 
   private static final String CHECK_TASK_QUEUE = "CHECK_CONNECTION";
   private static final String DISCOVER_TASK_QUEUE = "DISCOVER_SCHEMA";
-  private static final String UI_COMMANDS_TASK_QUEUE = "ui-commands-queue";
+  private static final String UI_COMMANDS_TASK_QUEUE = "ui_commands";
   private static final JobRunConfig JOB_RUN_CONFIG = new JobRunConfig()
       .withJobId(String.valueOf(JOB_ID))
       .withAttemptId((long) ATTEMPT_ID);
@@ -115,12 +116,10 @@ public class TemporalClientTest {
   private ConnectionManagerUtils connectionManagerUtils;
   private StreamResetRecordsHelper streamResetRecordsHelper;
   private Path workspaceRoot;
-  private String uiCommandsQueue;
 
   @BeforeEach
   void setup() throws IOException {
     workspaceRoot = Files.createTempDirectory(Path.of("/tmp"), "temporal_client_test");
-    uiCommandsQueue = "ui-commands-queue";
     logPath = workspaceRoot.resolve(String.valueOf(JOB_ID)).resolve(String.valueOf(ATTEMPT_ID)).resolve(DEFAULT_LOG_FILENAME);
     workflowClient = mock(WorkflowClient.class);
     when(workflowClient.getOptions()).thenReturn(WorkflowClientOptions.newBuilder().setNamespace(NAMESPACE).build());
@@ -138,7 +137,8 @@ void setup() throws IOException {
     connectionManagerUtils = spy(new ConnectionManagerUtils(workflowClientWrapped, metricClient));
     streamResetRecordsHelper = mock(StreamResetRecordsHelper.class);
     temporalClient =
-        spy(new TemporalClient(workspaceRoot, uiCommandsQueue, workflowClientWrapped, workflowServiceStubsWrapped, streamResetPersistence,
+        spy(new TemporalClient(workspaceRoot, new TemporalQueueConfiguration(), workflowClientWrapped, workflowServiceStubsWrapped,
+            streamResetPersistence,
             streamRefreshesRepository,
             connectionManagerUtils, streamResetRecordsHelper, mock(MetricClient.class), new TestClient(), scopedConfigurationService));
   }
@@ -155,7 +155,7 @@ void init() {
       final var metricClient = mock(MetricClient.class);
       final var scopedConfigurationService = mock(ScopedConfigurationService.class);
       temporalClient = spy(
-          new TemporalClient(workspaceRoot, uiCommandsQueue, new WorkflowClientWrapped(workflowClient, metricClient),
+          new TemporalClient(workspaceRoot, new TemporalQueueConfiguration(), new WorkflowClientWrapped(workflowClient, metricClient),
               new WorkflowServiceStubsWrapped(workflowServiceStubs, metricClient), streamResetPersistence, streamRefreshesRepository,
               mConnectionManagerUtils, streamResetRecordsHelper, metricClient, new TestClient(), scopedConfigurationService));
     }

airbyte-server/src/main/resources/application.yml

@@ -321,8 +321,6 @@ airbyte:
         max-jobs: ${MAX_FAILED_JOBS_IN_A_ROW_BEFORE_CONNECTION_DISABLE:30}
         max-fields-per-connection: ${MAX_FIELDS_PER_CONNECTION:20000}
   temporal:
-    queues:
-      ui-commands: ui_commands
   web-app:
     url: ${WEBAPP_URL:}
   workspace:

airbyte-workers/src/main/java/io/airbyte/workers/ApplicationInitializer.java

@@ -10,6 +10,7 @@
 import io.airbyte.commons.temporal.TemporalInitializationUtils;
 import io.airbyte.commons.temporal.TemporalJobType;
 import io.airbyte.commons.temporal.TemporalUtils;
+import io.airbyte.commons.temporal.config.TemporalQueueConfiguration;
 import io.airbyte.config.MaxWorkersConfig;
 import io.airbyte.micronaut.temporal.TemporalProxyHelper;
 import io.airbyte.workers.temporal.check.connection.CheckConnectionWorkflowImpl;
@@ -21,7 +22,6 @@
 import io.airbyte.workers.temporal.workflows.DiscoverCatalogAndAutoPropagateWorkflowImpl;
 import io.airbyte.workers.tracing.StorageObjectGetInterceptor;
 import io.airbyte.workers.tracing.TemporalSdkInterceptor;
-import io.micronaut.context.annotation.Property;
 import io.micronaut.context.annotation.Requires;
 import io.micronaut.context.annotation.Value;
 import io.micronaut.context.env.Environment;
@@ -106,6 +106,10 @@ public class ApplicationInitializer implements ApplicationEventListener<ServiceR
   private TemporalUtils temporalUtils;
   @Inject
   private WorkerFactory workerFactory;
+
+  @Inject
+  private TemporalQueueConfiguration temporalQueueConfiguration;
+
   @Value("${airbyte.data.sync.task-queue}")
   private String syncTaskQueue;
 
@@ -115,9 +119,6 @@ public class ApplicationInitializer implements ApplicationEventListener<ServiceR
   @Value("${airbyte.data.discover.task-queue}")
   private String discoverTaskQueue;
 
-  @Property(name = "airbyte.temporal.queues.ui-commands")
-  private String uiCommandsQueue;
-
   @Override
   public void onApplicationEvent(final ServiceReadyEvent event) {
     try {
@@ -178,7 +179,8 @@ private void registerWorkerFactory(final WorkerFactory workerFactory,
   }
 
   private void registerUiCommandsWorker(final WorkerFactory factory, final MaxWorkersConfig maxWorkersConfiguration) {
-    final Worker uiCommandsWorker = factory.newWorker(uiCommandsQueue, getWorkerOptions(maxWorkersConfiguration.getMaxCheckWorkers()));
+    final Worker uiCommandsWorker =
+        factory.newWorker(temporalQueueConfiguration.getUiCommandsQueue(), getWorkerOptions(maxWorkersConfiguration.getMaxCheckWorkers()));
     final WorkflowImplementationOptions workflowOptions = WorkflowImplementationOptions.newBuilder()
         .setFailWorkflowExceptionTypes(NonDeterministicException.class).build();
 

airbyte-workers/src/main/resources/application.yml

@@ -191,8 +191,6 @@ airbyte:
         client-secret: ${AB_AZURE_KEY_VAULT_CLIENT_SECRET:}
         tags: ${AB_AZURE_KEY_VAULT_TAGS:}
   temporal:
-    queues:
-      ui-commands: ui_commands
     worker:
       ports: ${TEMPORAL_WORKER_PORTS:}
   tracking: