def METHOD_NAME(
    executor_def: ExecutorDefinition,
) -> Tuple[JobDefinition, Callable[[str, int, int], Dict[str, Any]]]:
    # Induces failure state where among a series of dynamic steps induced from
    # upstream dynamic outputs, some of those steps fail during resource
    # initialization time. Since resource initialization is happening across
    # multiple processes, it is necessary to track the total number of
    # initializations that have already occurred within an external file
    # (count.pkl) so that state can be shared. In multiprocessing case, race
    # conditions are avoided by setting max concurrency to 1.
    @op(out=DynamicOut(), config_schema={"num_dynamic_steps": int})
    def source(context):
        for i in range(context.op_config["num_dynamic_steps"]):
            yield DynamicOutput(i, mapping_key=str(i))
    @resource(config_schema={"path": str, "allowed_initializations": int})
    def resource_for_dynamic_step(init_context):
        # Get the count of successful initializations. If it is already at the
        # allowed initialization number, fail.
        with open(os.path.join(init_context.resource_config["path"], "count.pkl"), "rb") as f:
            init_count = pickle.load(f)
            if init_count == init_context.resource_config["allowed_initializations"]:
                raise Exception("too many initializations.")
        # We have not yet reached the allowed initializaton number, allow
        # initialization to complete and add to total init number.
        with open(os.path.join(init_context.resource_config["path"], "count.pkl"), "wb") as f:
            init_count += 1
            pickle.dump(init_count, f)
        return None
    @op(required_resource_keys={"foo"})
    def mapped_op(x):
        pass
    @op
    def consumer(x):
        pass
    @job(resource_defs={"foo": resource_for_dynamic_step}, executor_def=executor_def)
    def the_job():
        consumer(source().map(mapped_op).collect())
    return (
        the_job,
        lambda temp_dir, init_count, dynamic_steps: {
            "resources": {
                "foo": {"config": {"path": temp_dir, "allowed_initializations": init_count}}
            },
            "ops": {"source": {"config": {"num_dynamic_steps": dynamic_steps}}},
        },
    )