[adapters] Fix now connector in multihost.

blp · blp · commit 0343fd6f67cf · 2026-02-27T20:13:04.000Z
The clock connector requests steps instead of queueing data, but we were
ignoring that in multihost.

Also, connect the clock only on the first host in a multihost cluster.

This fixes a Python integration test for now in multihost, so enable that
test.

Signed-off-by: Ben Pfaff &lt;blp@feldera.com&gt;
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -3737,7 +3737,14 @@ impl StepTrigger {
         let result = if let Some(request) = &coordination_request {
             match request.action {
                 StepAction::Step if request.step >= step => Some(Action::Step),
-                StepAction::Trigger if request.step >= step => None,
+                StepAction::Trigger if request.step >= step => {
+                    if self.controller.status.unset_step_requested() {
+                        // The `clock` connector requests steps explicitly.
+                        Some(Action::Step)
+                    } else {
+                        None
+                    }
+                }
                 _ => Some(Action::Park(None)),
             }
         } else if replaying
@@ -5143,7 +5150,9 @@ impl ControllerInner {
             controller.status.set_state(PipelineState::Terminated);
         })?;
 
-        let _ = controller.connect_input("now", &now_endpoint_config(&config), None);
+        if controller.workers.start == 0 {
+            let _ = controller.connect_input("now", &now_endpoint_config(&config), None);
+        }
 
         let backpressure_thread =
             BackpressureThread::new(controller.clone(), backpressure_thread_parker);
diff --git a/python/tests/platform/test_nowstream.py b/python/tests/platform/test_nowstream.py
@@ -7,7 +7,6 @@
     unique_pipeline_name,
     FELDERA_TEST_NUM_WORKERS,
     FELDERA_TEST_NUM_HOSTS,
-    single_host_only,
 )
 from tests import TEST_CLIENT
 from feldera.enums import PipelineStatus
@@ -20,7 +19,6 @@ def get_result(pipeline) -> str:
 
 
 class TestNowStream(unittest.TestCase):
-    @single_host_only
     def test_nowstream(self):
         """
         Test the now() function:
