def METHOD_NAME(job_fn, config_fn):
    num_dynamic_steps = 3
    with TemporaryDirectory() as temp_dir:
        with instance_for_test(temp_dir=temp_dir) as instance:
            _write_blank_count(temp_dir)
            result = execute_job(
                reconstructable(job_fn),
                instance,
                run_config=config_fn(
                    temp_dir, 1, num_dynamic_steps
                ),  # Only allow one dynamic step to pass
            )
            assert not result.success
            assert len(result.get_step_success_events()) == 2
            assert _regexes_match(
                [r"source", r"mapped_op\[\d\]"],
                [event.step_key for event in result.get_step_success_events()],
            )
            assert len(result.get_failed_step_keys()) == 2
            assert _regexes_match(
                [r"mapped_op\[\d\]", r"mapped_op\[\d\]"],
                list(result.get_failed_step_keys()),
            )
            _write_blank_count(temp_dir)
            retry_result = execute_job(
                reconstructable(job_fn),
                instance,
                run_config=config_fn(
                    temp_dir, num_dynamic_steps, num_dynamic_steps
                ),  # Allows all dynamic steps to pass
                reexecution_options=ReexecutionOptions.from_failure(
                    run_id=result.run_id, instance=instance
                ),
            )
            assert retry_result.success
            assert len(retry_result.get_step_success_events()) == 3
            assert _regexes_match(
                [r"mapped_op\[\d\]", r"mapped_op\[\d\]", "consumer"],
                [event.step_key for event in retry_result.get_step_success_events()],
            )