code: def make_notify_execution_state_change_fn(
      node_uid: task_lib.NodeUid,
  ) -> Callable[
      [
          Optional[metadata_store_pb2.Execution],
          metadata_store_pb2.Execution,
      ],
      None,
  ]:
      """Returns a on_commit callback for use with mlmd_execution_atomic_op.
  
      Args:
        node_uid: The NodeUid for the node whose execution is being updated.
  
      Returns:
        An on_commit callback for use with mlmd_execution_atomic_op. The callback
        sends an ExecutionStateChange notification if the execution state changed.
      """
  
      def on_commit(
          pre_commit_execution: Optional[metadata_store_pb2.Execution],
          post_commit_execution: metadata_store_pb2.Execution,
      ) -> None:
          pre_commit_execution_state = None
          if pre_commit_execution:
              pre_commit_execution_state = (
                  pre_commit_execution.last_known_state
              )
          if (
              pre_commit_execution_state
              == post_commit_execution.last_known_state
          ):
              return
          notify(
              ExecutionStateChange(
                  execution=post_commit_execution,
                  node_uid=node_uid,
                  old_state=pre_commit_execution_state,
                  new_state=post_commit_execution.last_known_state,
              )
          )
  
      return on_commit
  
imports: from concurrent import futures
  import contextlib
  import dataclasses
  import queue
  import threading
  from typing import Any, Callable, List, Optional, Union
  from absl import logging
  from tfx.orchestration.experimental.core import task
  from tfx.utils import status
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 6 for calling a global function with example usage: make_notify_execution_state_change_fn(node_uid) and returns: on_commit