Michail Melonas
01/19/2022, 3:10 PMKevin Kho
flow.set_reference_tasks
to set the reference task for the FlowMichail Melonas
01/20/2022, 7:18 AMTask
, e.g.:
class MyTask(Task):
def run(self):
from random import random
if random() > 0.5: return "Hello, World!"
else:
# set the state of this Task to Failed (Finished)
pass
Kevin Kho
raise FAIL
where FAIL is the Prefect signalMichail Melonas
01/20/2022, 7:27 AMfrom prefect.engine.signals import FAIL
class MyTask(Task):
def run(self):
from random import random
if random() > 0.5: return "Hello, World!"
else: raise FAIL()
Kevin Kho
Michail Melonas
02/16/2022, 1:58 PMis_finished
method) are not called. However, when using the “CANCEL” button the behaviour is as expected.Kevin Kho
def mystatehandler(....)
if isinstance(new_state, Cancelled):
...
works but I wasn’t sureMichail Melonas
02/16/2022, 2:38 PMdef kill_keda_job_callback(flow, old_state, new_state):
if new_state.is_finished():
pass
return new_state
def delete_queues_callback(flow, old_state, new_state):
if new_state.is_finished():
<http://logger.info|logger.info>(f"Deleting {flow.work_queue_name} and {flow.callback_queue_name}")
with Connection() as (_, channel):
channel.queue_delete(queue=flow.work_queue_name)
channel.queue_delete(queue=flow.callback_queue_name)
return new_state
flow.state_handlers = [kill_keda_job_callback, delete_queues_callback]
Kevin Kho
Michail Melonas
02/16/2022, 2:43 PMException
or prefect.engine.signals.FAIL
gets raised during task execution, or if the “CANCEL” button is employed.Kevin Kho
def mystatehandler(flow, old_state, new_state):
if isinstance(new_state, Cancelled):
and also:
def kill_keda_job_callback(flow, old_state, new_state):
if new_state.is_finished():
gets triggered because of the finished state.
When we set the state to Failed in the UI, there are two possible outcomes. The first is if there are still downstream tasks, the Flow gets confused and there are Client errors. The second is if you mark as FAILED during a terminal task. In this case, that last task may still succeed causing the state to succeed. State handlers for is_failed()
are not hit in this case
This is because setting the state in the UI and using the Cancel button take two code paths. Setting the state in the UI is a direct modification of state in the Prefect database. It does not touch the Flow process. Using the Cancel button goes into the Flow process and tries to terminate it. Because it is done in the Flow process, the Flow can still run the state handlers while exiting.
So the summary is it’s not by design but you are right that Cancel works differently and setting state in the UI does notMichail Melonas
02/17/2022, 10:13 AMKevin Kho