code: def task_finished(
      self,
      step_name,
      flow,
      graph,
      is_task_ok,
      retry_count,
      max_user_code_retries,
  ):
      if not is_task_ok:
          # The task finished with an exception - execution won't
          # continue so no need to do anything here.
          return
  
      # For `foreach`s, we need to dump the cardinality of the fanout
      # into AWS DynamoDb so that AWS Step Functions can properly configure
      # the Map job, in the absence of any better message passing feature
      # between the states.
      if graph[step_name].type == "foreach":
          # Since we can't generate the full path spec within AWS Step
          # Function DynamoDb Get task, we will just key by task id for now.
          # Also, we can afford to set the ttl only once for the key in AWS
          # DynamoDB here. AWS Step Functions can execute for up to a year
          # and execution history is available for 90 days after the
          # execution.
          self._save_foreach_cardinality(
              os.environ["AWS_BATCH_JOB_ID"],
              flow._foreach_num_splits,
              self._ttl(),
          )
      # The parent task ids need to be available in a foreach join so that
      # we can construct the input path. Unfortunately, while AWS Step
      # Function provides access to an array of parent task ids, we can't
      # make use of them since AWS Batch job spec only accepts strings. We
      # instead write the task ids from the parent task to DynamoDb and read
      # it back in the foreach join
      elif graph[step_name].is_inside_foreach and any(
          graph[n].type == "join"
          and graph[graph[n].split_parents[-1]].type == "foreach"
          for n in graph[step_name].out_funcs
      ):
          self._save_parent_task_id_for_foreach_join(
              os.environ[
                  "METAFLOW_SPLIT_PARENT_TASK_ID_FOR_FOREACH_JOIN"
              ],
              os.environ["AWS_BATCH_JOB_ID"],
          )
  
file_path: metaflow/metaflow/plugins/aws/step_functions/step_functions_decorator.py
imports: import os
  import json
  import time
  from metaflow.decorators import StepDecorator
  from metaflow.metadata import MetaDatum
  from .dynamo_db_client import DynamoDbClient
label_desc: Use image node_id 2 for calling the StepFunctionsInternalDecorator obj's underlying member method code with example usage: obj.task_finished(step_name, flow, graph, is_task_ok, retry_count, max_user_code_retries) without return types