code: def _set_node_execution_options(
      node: pipeline_pb2.PipelineNode,
      tfx_node: base_node.BaseNode,
      pipeline_ctx: compiler_context.PipelineContext,
      enable_cache: bool,
  ):
      """Compiles and sets NodeExecutionOptions of a pipeline node."""
      options_proto = pipeline_pb2.NodeExecutionOptions()
      options_proto.caching_options.enable_cache = enable_cache
      options_py = tfx_node.node_execution_options
      if options_py:
          assert isinstance(
              options_py, execution_options_utils.NodeExecutionOptions
          )
          if (
              options_py.trigger_strategy or options_py.success_optional
          ) and not pipeline_ctx.is_sync_mode:
              raise ValueError(
                  "Node level triggering strategies and success "
                  "optionality are only used in SYNC pipelines."
              )
          if (
              options_py.trigger_strategy
              == pipeline_pb2.NodeExecutionOptions.LIFETIME_END_WHEN_SUBGRAPH_CANNOT_PROGRESS
              and not options_py.lifetime_start
          ):
              raise ValueError(
                  f"Node {node.node_info.id} has the trigger strategy"
                  " LIFETIME_END_WHEN_SUBGRAPH_CANNOT_PROGRESS set but no"
                  " lifetime_start. In order to use the trigger strategy the node"
                  " must have a lifetime_start."
              )
          options_proto.strategy = options_py.trigger_strategy
          options_proto.node_success_optional = (
              options_py.success_optional
          )
          if options_py.max_execution_retries is not None:
              options_proto.max_execution_retries = (
                  options_py.max_execution_retries
              )
          options_proto.execution_timeout_sec = (
              options_py.execution_timeout_sec
          )
          options_proto.run_mode = (
              options_py._run_mode
          )  # pylint: disable=protected-access
          options_proto.reset_stateful_working_dir = (
              options_py.reset_stateful_working_dir
          )
          if options_py.lifetime_start:
              options_proto.resource_lifetime.lifetime_start = (
                  options_py.lifetime_start
              )
      node.execution_options.CopyFrom(options_proto)
  
      # TODO: b/310726801 - We should throw an error if this is an invalid
      # configuration.
      if pipeline_ctx.is_async_mode:
          input_triggers = (
              node.execution_options.async_trigger.input_triggers
          )
          for input_key, input_channel in tfx_node.inputs.items():
              if isinstance(
                  input_channel.input_trigger, channel_types.NoTrigger
              ):
                  input_triggers[input_key].no_trigger = True
              if isinstance(
                  input_channel.input_trigger,
                  channel_types.TriggerByProperty,
              ):
                  input_triggers[
                      input_key
                  ].trigger_by_property.property_keys.extend(
                      input_channel.input_trigger.property_keys
                  )
  
imports: import itertools
  from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, Type, cast
  from tfx import types
  from tfx.dsl.compiler import compiler_context
  from tfx.dsl.compiler import compiler_utils
  from tfx.dsl.compiler import constants
  from tfx.dsl.compiler import node_inputs_compiler
  from tfx.dsl.components.base import base_component
  from tfx.dsl.components.base import base_driver
  from tfx.dsl.components.base import base_node
  from tfx.dsl.experimental.node_execution_options import utils
  from tfx.dsl.placeholder import placeholder
  from tfx.orchestration import data_types
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import pipeline
  from tfx.proto.orchestration import executable_spec_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import channel
  from tfx.types import channel_utils
  from tfx.utils import deprecation_utils
  from tfx.utils import name_utils
label_desc: Use image node_id 8 for calling a global function with example usage: _set_node_execution_options(node, tfx_node, pipeline_ctx, enable_cache) without return types