chore: disable check and discover all task queue routing (#14427)

airbytehq · Oct 22, 2024 · 1564e9d · 1564e9d
1 parent 313766c
commit 1564e9d
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 0 deletions.
diff --git a/...-commons-temporal/src/main/java/io/airbyte/commons/temporal/scheduling/RouterService.java b/...-commons-temporal/src/main/java/io/airbyte/commons/temporal/scheduling/RouterService.java
@@ -90,6 +90,12 @@ public String getTaskQueueForWorkspace(final UUID workspaceId, final TemporalJob
       throw new RuntimeException("Jobtype not expected to call - getTaskQueueForWorkspace - " + jobType);
     }
 
+    // Feature flag to disable the data-plane routing of temporal queues
+    if (!featureFlagClient.boolVariation(UseRouteToTaskRouting.INSTANCE,
+        new Multi(List.of(new Workspace(workspaceId))))) {
+      return jobType.name();
+    }
+
     final Geography geography = workspaceService.getGeographyForWorkspace(workspaceId);
     if (featureFlagClient.boolVariation(ShouldRunOnGkeDataplane.INSTANCE, new Workspace(workspaceId))) {
       // Routing logic to route dataplane jobs to expanded dataplane

diff --git a/airbyte-workers/src/test/java/io/airbyte/commons/temporal/scheduling/RouterServiceTest.java b/airbyte-workers/src/test/java/io/airbyte/commons/temporal/scheduling/RouterServiceTest.java
@@ -133,6 +133,7 @@ void testGetTaskQueueBehindExpandedFlag() throws IOException, ConfigNotFoundExce
 
   @Test
   void testGetWorkspaceTaskQueue() throws IOException, ConfigNotFoundException {
+    Mockito.when(mockFeatureFlagClient.boolVariation(eq(UseRouteToTaskRouting.INSTANCE), any())).thenReturn(true);
     Mockito.when(mockFeatureFlagClient.boolVariation(ShouldRunOnGkeDataplane.INSTANCE, new Workspace(WORKSPACE_ID))).thenReturn(false);
 
     Mockito.when(mWorkspaceService.getGeographyForWorkspace(WORKSPACE_ID)).thenReturn(Geography.AUTO);
@@ -147,6 +148,7 @@ void testGetWorkspaceTaskQueue() throws IOException, ConfigNotFoundException {
 
   @Test
   void testGetWorkspaceTaskQueueBehindFlag() throws IOException, ConfigNotFoundException {
+    Mockito.when(mockFeatureFlagClient.boolVariation(eq(UseRouteToTaskRouting.INSTANCE), any())).thenReturn(true);
     Mockito.when(mockFeatureFlagClient.boolVariation(ShouldRunOnGkeDataplane.INSTANCE, new Workspace(WORKSPACE_ID))).thenReturn(true);
 
     Mockito.when(mWorkspaceService.getGeographyForWorkspace(WORKSPACE_ID)).thenReturn(Geography.AUTO);
@@ -161,6 +163,7 @@ void testGetWorkspaceTaskQueueBehindFlag() throws IOException, ConfigNotFoundExc
 
   @Test
   void testGetWorkspaceOnExpandedTaskQueue() throws IOException, ConfigNotFoundException {
+    Mockito.when(mockFeatureFlagClient.boolVariation(eq(UseRouteToTaskRouting.INSTANCE), any())).thenReturn(true);
     Mockito.when(mockFeatureFlagClient.boolVariation(ShouldRunOnGkeDataplane.INSTANCE, new Workspace(WORKSPACE_ID))).thenReturn(true);
     Mockito.when(mockFeatureFlagClient.boolVariation(ShouldRunOnExpandedGkeDataplane.INSTANCE, new Workspace(WORKSPACE_ID))).thenReturn(true);