Great question. In Prefect 3.x, if you want a task to “crash” (not just fail) to reproduce bad logging behavior, you’ll want to terminate the executing process abruptly so logs don’t flush. The safest way to do this without taking down your whole flow is to run tasks in a subprocess and then hard-exit from inside the task. Quick clarifying question: - Which task runner are you using (Concurrent/ThreadPool, ProcessPool, Dask, Kubernetes)? - Do you specifically want to lose the task’s logs (i.e., simulate a logging pipeline that drops messages on crash)? In the meantime, here are minimal recipes you can use right now. Option A: Hard crash a task subprocess (best for reproducing lost logs) - Use ProcessPoolTaskRunner so each task runs in its own process - Inside the task, call os._exit(1) or send SIGKILL. This bypasses Python cleanup and logging flushes.

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner
from prefect.logging.loggers import get_run_logger
import os
import signal

@task
def log_then_die_hard():
    logger = get_run_logger()
    <http://logger.info|logger.info>("This log is likely to be lost if the process dies immediately")
    # Hard terminate the process without flushing stdio/log handlers:
    os._exit(1)
    # Alternatively (even harsher): os.kill(os.getpid(), signal.SIGKILL)

@flow(task_runner=ProcessPoolTaskRunner())
def demo_crash():
    f = log_then_die_hard.submit()
    try:
        f.result()
    except Exception as e:
        print(f"Flow saw exception: {type(e).__name__}: {e}")

if __name__ == "__main__":
    demo_crash()

What you’ll see: - The task will end in a Crashed state. - The “This log…” line often won’t appear in Prefect logs (bad logging reproduced). - The flow catches an exception when retrieving the result, but the key is the Crashed state and missing logs. Option B: Simulate a segfault (also drops logs)

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner
from prefect.logging.loggers import get_run_logger
import ctypes

@task
def segfault():
    logger = get_run_logger()
    logger.warning("About to segfault; this may not be logged")
    ctypes.string_at(0)  # segfault

@flow(task_runner=ProcessPoolTaskRunner())
def demo_segfault():
    f = segfault.submit()
    try:
        f.result()
    except Exception as e:
        print(f"Flow saw: {type(e).__name__}: {e}")

if __name__ == "__main__":
    demo_segfault()

Notes: - With ThreadPool/ConcurrentTaskRunner, a hard kill like os._exit(1) or SIGKILL will take down the whole Python process (flow/worker), not just the task. That’s why ProcessPoolTaskRunner is ideal for isolating a task crash. - If you just need a “crashed” state but still want logs to show up, use a softer signal like SIGTERM or raise Prefect’s TerminationSignal. These usually allow logging to flush: - Soft crash (often preserves logs): os.kill(os.getpid(), signal.SIGTERM) - Prefect-specific crash: raise prefect.exceptions.TerminationSignal("Test crash") If you can share your runner (e.g., Dask/Kubernetes) and where you expect logs to appear (Cloud UI, stdout, external collector), I can tailor the snippet to match your setup.