code: def test_get_unprocessed_inputs_with_retry_limit(self):
      with self._mlmd_connection as m:
          # Fake one output of self._example_gen.
          otu.fake_upstream_node_run(
              m,
              self._example_gen,
              fake_result="Tflex rocks.",
              tmp_path=self.create_tempfile().full_path,
          )
          contexts = m.store.get_contexts()
          artifact_types = m.store.get_artifact_types()
          artifacts = artifact_utils.deserialize_artifacts(
              artifact_types[0], m.store.get_artifacts()
          )
          input_and_param = task_gen_utils.InputAndParam(
              input_artifacts={"examples": artifacts}
          )
          resolved_info_for_transform = task_gen_utils.ResolvedInfo(
              contexts=contexts,
              input_and_params=[input_and_param],
          )
  
          # Set the maximum retry of self._transform to 2.
          self._transform.execution_options.max_execution_retries = 2
  
          # Simulate that self._transform failed the first time.
          execution = otu.fake_start_node_with_handle(
              m,
              self._transform,
              input_artifacts={"examples": artifacts},
          )
          otu.fake_finish_node_with_handle(
              m, self._transform, execution.id, success=False
          )
          self.assertIsNone(input_and_param.exec_properties)
          unprocessed_inputs = task_gen_utils.get_unprocessed_inputs(
              m, resolved_info_for_transform, self._transform
          )
          self.assertIsNotNone(unprocessed_inputs[0].exec_properties)
          self.assertLen(unprocessed_inputs, 1)
  
          # Simulate that self._transform retry twice.
          execution = otu.fake_start_node_with_handle(
              m,
              self._transform,
              input_artifacts={"examples": artifacts},
          )
          otu.fake_finish_node_with_handle(
              m, self._transform, execution.id, success=False
          )
          execution = otu.fake_start_node_with_handle(
              m,
              self._transform,
              input_artifacts={"examples": artifacts},
          )
          otu.fake_finish_node_with_handle(
              m, self._transform, execution.id, success=False
          )
  
          # Since self._transform has retried twice, we won't try it again, so the
          # unprocessed_inputs is empty.
          unprocessed_inputs = task_gen_utils.get_unprocessed_inputs(
              m, resolved_info_for_transform, self._transform
          )
          self.assertEmpty(unprocessed_inputs)
  
imports: import os
  import time
  from unittest import mock
  import uuid
  from absl.testing import parameterized
  import tensorflow
  from tfx import types
  from tfx import version
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_async_pipeline
  from tfx.orchestration.experimental.core.testing import test_dynamic_exec_properties_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import placeholder_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.utils import status
  from tfx.utils import test_case_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 17 for calling the TaskGenUtilsTest obj's underlying member method code with example usage: obj.test_get_unprocessed_inputs_with_retry_limit() without return types