Target wait warning to more specific circumstances (#16969)

src/prefect/task_engine.py

@@ -249,7 +249,7 @@ def log_finished_message(self) -> None:
         display_state = repr(self.state) if PREFECT_DEBUG_MODE else str(self.state)
         level = logging.INFO if self.state.is_completed() else logging.ERROR
         msg = f"Finished in state {display_state}"
-        if self.state.is_pending():
+        if self.state.is_pending() and self.state.name != "NotReady":
             msg += (
                 "\nPlease wait for all submitted tasks to complete"
                 " before exiting your flow by calling `.wait()` on the "

tests/test_task_engine.py

@@ -35,7 +35,7 @@
 from prefect.results import ResultRecord, ResultStore
 from prefect.server.schemas.core import ConcurrencyLimitV2
 from prefect.settings import PREFECT_TASK_DEFAULT_RETRIES, temporary_settings
-from prefect.states import Completed, Running, State
+from prefect.states import Completed, Pending, Running, State
 from prefect.task_engine import (
     AsyncTaskRunEngine,
     SyncTaskRunEngine,
@@ -101,6 +101,37 @@ async def test_set_task_run_state_duplicated_timestamp(self):
             assert new_state == completed_state
             assert new_state.timestamp > running_state.timestamp
 
+    def test_logs_message_when_submitted_tasks_end_in_pending(self, caplog):
+        """
+        If submitted tasks aren't waited on before a flow exits, they may fail to run
+        because they're transition from PENDING to RUNNING is denied. This test ensures
+        that a message is logged when this happens.
+        """
+        engine = SyncTaskRunEngine(task=foo)
+        with engine.initialize_run():
+            assert engine.state.is_pending()
+
+        assert (
+            "Please wait for all submitted tasks to complete before exiting your flow"
+            in caplog.text
+        )
+
+    def test_doesnt_log_message_when_submitted_tasks_end_in_not_ready(self, caplog):
+        """
+        Regression test for tasks that didn't run because of upstream issues, not because of
+        a lack of wait call. See https://github.com/PrefectHQ/prefect/issues/16848
+        """
+
+        engine = SyncTaskRunEngine(task=foo)
+        with engine.initialize_run():
+            assert engine.state.is_pending()
+            engine.set_state(Pending(name="NotReady"))
+
+        assert (
+            "Please wait for all submitted tasks to complete before exiting your flow"
+            not in caplog.text
+        )
+
 
 class TestAsyncTaskRunEngine:
     async def test_basic_init(self):

tests/test_tasks.py

@@ -910,35 +910,6 @@ def my_flow():
         with pytest.raises(ValueError, match="deadlock"):
             my_flow()
 
-    @pytest.mark.skip(
-        reason="This test is not compatible with the current state of client side task orchestration"
-    )
-    def test_logs_message_when_submitted_tasks_end_in_pending(self, caplog):
-        """
-        If submitted tasks aren't waited on before a flow exits, they may fail to run
-        because they're transition from PENDING to RUNNING is denied. This test ensures
-        that a message is logged when this happens.
-        """
-
-        @task
-        def find_palindromes():
-            """This is a computationally expensive task that never ends,
-            allowing the flow to exit before the task is completed."""
-            num = 10
-            while True:
-                _ = str(num) == str(num)[::-1]
-                num += 1
-
-        @flow
-        def test_flow():
-            find_palindromes.submit()
-
-        test_flow()
-        assert (
-            "Please wait for all submitted tasks to complete before exiting your flow"
-            in caplog.text
-        )
-
 
 class TestTaskStates:
     @pytest.mark.parametrize("error", [ValueError("Hello"), None])