Don't bother with balancer when not using sticky queues

Author: Sushisource

core/src/pollers/poll_buffer.rs

@@ -53,7 +53,7 @@ pub(crate) struct LongPollBuffer<T, SK: SlotKind> {
 }
 
 pub(crate) struct WorkflowTaskOptions {
-    pub(crate) wft_poller_shared: Arc<WFTPollerShared>,
+    pub(crate) wft_poller_shared: Option<Arc<WFTPollerShared>>,
 }
 
 pub(crate) struct ActivityTaskOptions {
@@ -77,27 +77,27 @@ impl LongPollBuffer<PollWorkflowTaskQueueResponse, WorkflowSlotKind> {
     ) -> Self {
         let is_sticky = sticky_queue.is_some();
         let poll_scaler = PollScaler::new(poller_behavior, num_pollers_handler, shutdown.clone());
-        if is_sticky {
-            options
-                .wft_poller_shared
-                .set_sticky_active(poll_scaler.active_rx.clone());
-        } else {
-            options
-                .wft_poller_shared
-                .set_non_sticky_active(poll_scaler.active_rx.clone());
-        };
-        let shared = options.wft_poller_shared.clone();
-        let pre_permit_delay = Some(move || {
-            let shared = shared.clone();
-            async move {
-                shared.wait_if_needed(is_sticky).await;
+        if let Some(wftps) = options.wft_poller_shared.as_ref() {
+            if is_sticky {
+                wftps.set_sticky_active(poll_scaler.active_rx.clone());
+            } else {
+                wftps.set_non_sticky_active(poll_scaler.active_rx.clone());
+            };
+        }
+        let pre_permit_delay = options.wft_poller_shared.clone().map(|wftps| {
+            move || {
+                let shared = wftps.clone();
+                async move {
+                    shared.wait_if_needed(is_sticky).await;
+                }
             }
         });
-        let post_poll_fn = Some(move |t: &PollWorkflowTaskQueueResponse| {
-            if is_sticky {
-                options
-                    .wft_poller_shared
-                    .record_sticky_backlog(t.backlog_count_hint as usize)
+
+        let post_poll_fn = options.wft_poller_shared.clone().map(|wftps| {
+            move |t: &PollWorkflowTaskQueueResponse| {
+                if is_sticky {
+                    wftps.record_sticky_backlog(t.backlog_count_hint as usize)
+                }
             }
         });
         let no_retry = if matches!(poller_behavior, PollerBehavior::Autoscaling { .. }) {
@@ -724,7 +724,7 @@ mod tests {
             CancellationToken::new(),
             None::<fn(usize)>,
             WorkflowTaskOptions {
-                wft_poller_shared: Arc::new(WFTPollerShared::new()),
+                wft_poller_shared: Some(Arc::new(WFTPollerShared::new())),
             },
         );
 

core/src/worker/workflow/wft_poller.rs

@@ -34,7 +34,11 @@ pub(crate) fn make_wft_poller(
 > + Sized
 + 'static {
     let wft_metrics = metrics.with_new_attrs([workflow_poller()]);
-    let wft_poller_shared = Arc::new(WFTPollerShared::new());
+    let wft_poller_shared = if sticky_queue_name.is_some() {
+        Some(Arc::new(WFTPollerShared::new()))
+    } else {
+        None
+    };
     let wf_task_poll_buffer = LongPollBuffer::new_workflow_task(
         client.clone(),
         config.task_queue.clone(),