code: def extract_stateful_function(
      user_defined_function_proto,
      runtime_context: RuntimeContext,
      keyed_state_backend,
      operator_state_store,
  ):
      from pyflink.fn_execution import flink_fn_execution_pb2
  
      func_type = user_defined_function_proto.function_type
      user_defined_func = pickle.loads(
          user_defined_function_proto.payload
      )
      has_side_output = user_defined_function_proto.has_side_output
      internal_timer_service = InternalTimerServiceImpl(
          keyed_state_backend
      )
  
      def state_key_selector(normal_data):
          return Row(normal_data[0])
  
      def user_key_selector(normal_data):
          return normal_data[0]
  
      def input_selector(normal_data):
          return normal_data[1]
  
      UserDefinedDataStreamFunction = (
          flink_fn_execution_pb2.UserDefinedDataStreamFunction
      )
      if func_type in (
          UserDefinedDataStreamFunction.KEYED_PROCESS,
          UserDefinedDataStreamFunction.KEYED_CO_PROCESS,
      ):
          timer_service = TimerServiceImpl(internal_timer_service)
          ctx = InternalKeyedProcessFunctionContext(timer_service)
          on_timer_ctx = InternalKeyedProcessFunctionOnTimerContext(
              timer_service
          )
          process_function = user_defined_func
          internal_timer_service.set_namespace_serializer(
              VoidNamespaceSerializer()
          )
  
          def open_func():
              if hasattr(process_function, "open"):
                  process_function.open(runtime_context)
  
          def close_func():
              if hasattr(process_function, "close"):
                  process_function.close()
  
          def on_event_time(timestamp: int, key, namespace):
              keyed_state_backend.set_current_key(key)
              return _on_timer(TimeDomain.EVENT_TIME, timestamp, key)
  
          def on_processing_time(timestamp: int, key, namespace):
              keyed_state_backend.set_current_key(key)
              return _on_timer(
                  TimeDomain.PROCESSING_TIME, timestamp, key
              )
  
          def _on_timer(time_domain: TimeDomain, timestamp: int, key):
              user_current_key = user_key_selector(key)
  
              on_timer_ctx.set_timestamp(timestamp)
              on_timer_ctx.set_current_key(user_current_key)
              on_timer_ctx.set_time_domain(time_domain)
  
              return process_function.on_timer(timestamp, on_timer_ctx)
  
          if func_type == UserDefinedDataStreamFunction.KEYED_PROCESS:
  
              def process_element(normal_data, timestamp: int):
                  ctx.set_timestamp(timestamp)
                  ctx.set_current_key(user_key_selector(normal_data))
                  keyed_state_backend.set_current_key(
                      state_key_selector(normal_data)
                  )
                  return process_function.process_element(
                      input_selector(normal_data), ctx
                  )
  
          elif (
              func_type
              == UserDefinedDataStreamFunction.KEYED_CO_PROCESS
          ):
  
              def process_element(normal_data, timestamp: int):
                  is_left = normal_data[0]
                  if is_left:
                      user_input = normal_data[1]
                  else:
                      user_input = normal_data[2]
  
                  ctx.set_timestamp(timestamp)
                  current_user_key = user_key_selector(user_input)
                  ctx.set_current_key(current_user_key)
                  on_timer_ctx.set_current_key(current_user_key)
                  keyed_state_backend.set_current_key(
                      state_key_selector(user_input)
                  )
  
                  if is_left:
                      return process_function.process_element1(
                          input_selector(user_input), ctx
                      )
                  else:
                      return process_function.process_element2(
                          input_selector(user_input), ctx
                      )
  
          else:
              raise Exception(
                  "Unsupported func_type: " + str(func_type)
              )
  
      elif (
          func_type
          == UserDefinedDataStreamFunction.KEYED_CO_BROADCAST_PROCESS
      ):
          timer_service = TimerServiceImpl(internal_timer_service)
          ctx = InternalKeyedBroadcastProcessFunctionContext(
              timer_service, operator_state_store
          )
          read_only_ctx = (
              InternalKeyedBroadcastProcessFunctionReadOnlyContext(
                  timer_service, operator_state_store
              )
          )
          on_timer_ctx = (
              InternalKeyedBroadcastProcessFunctionOnTimerContext(
                  timer_service, operator_state_store
              )
          )
          internal_timer_service.set_namespace_serializer(
              VoidNamespaceSerializer()
          )
  
          def open_func():
              if hasattr(user_defined_func, "open"):
                  user_defined_func.open(runtime_context)
  
          def close_func():
              if hasattr(user_defined_func, "close"):
                  user_defined_func.close()
  
          def on_event_time(timestamp: int, key, namespace):
              keyed_state_backend.set_current_key(key)
              return _on_timer(TimeDomain.EVENT_TIME, timestamp, key)
  
          def on_processing_time(timestamp: int, key, namespace):
              keyed_state_backend.set_current_key(key)
              return _on_timer(
                  TimeDomain.PROCESSING_TIME, timestamp, key
              )
  
          def _on_timer(time_domain: TimeDomain, timestamp: int, key):
              user_current_key = user_key_selector(key)
  
              on_timer_ctx.set_timestamp(timestamp)
              on_timer_ctx.set_current_key(user_current_key)
              on_timer_ctx.set_time_domain(time_domain)
  
              return user_defined_func.on_timer(timestamp, on_timer_ctx)
  
          def process_element(normal_data, timestamp):
              ctx.set_timestamp(timestamp)
              read_only_ctx.set_timestamp(timestamp)
  
              if normal_data[0]:
                  data = normal_data[1]
                  read_only_ctx.set_current_key(user_key_selector(data))
                  keyed_state_backend.set_current_key(
                      state_key_selector(data)
                  )
                  return user_defined_func.process_element(
                      input_selector(data), read_only_ctx
                  )
              else:
                  return user_defined_func.process_broadcast_element(
                      normal_data[2], ctx
                  )
  
      elif func_type == UserDefinedDataStreamFunction.WINDOW:
          window_operation_descriptor = (
              user_defined_func
          )  # type: WindowOperationDescriptor
          window_assigner = window_operation_descriptor.assigner
          window_trigger = window_operation_descriptor.trigger
          allowed_lateness = (
              window_operation_descriptor.allowed_lateness
          )
          late_data_output_tag = (
              window_operation_descriptor.late_data_output_tag
          )
          window_state_descriptor = (
              window_operation_descriptor.window_state_descriptor
          )
          internal_window_function = (
              window_operation_descriptor.internal_window_function
          )
          window_serializer = (
              window_operation_descriptor.window_serializer
          )
          window_coder = window_serializer._get_coder()
          keyed_state_backend.namespace_coder = window_coder
          keyed_state_backend._namespace_coder_impl = (
              window_coder.get_impl()
          )
          window_operator = WindowOperator(
              window_assigner,
              keyed_state_backend,
              user_key_selector,
              window_state_descriptor,
              internal_window_function,
              window_trigger,
              allowed_lateness,
              late_data_output_tag,
          )
          internal_timer_service.set_namespace_serializer(
              window_serializer
          )
  
          def open_func():
              window_operator.open(
                  runtime_context, internal_timer_service
              )
  
          def close_func():
              window_operator.close()
  
          def process_element(normal_data, timestamp: int):
              keyed_state_backend.set_current_key(
                  state_key_selector(normal_data)
              )
              return window_operator.process_element(
                  input_selector(normal_data), timestamp
              )
  
          def on_event_time(timestamp: int, key, namespace):
              keyed_state_backend.set_current_key(key)
              return window_operator.on_event_time(
                  timestamp, key, namespace
              )
  
          def on_processing_time(timestamp: int, key, namespace):
              keyed_state_backend.set_current_key(key)
              return window_operator.on_processing_time(
                  timestamp, key, namespace
              )
  
      else:
          raise Exception(
              "Unsupported function_type: " + str(func_type)
          )
  
      input_handler = RunnerInputHandler(
          internal_timer_service, process_element, has_side_output
      )
      process_element_func = input_handler.process_element
  
      timer_handler = TimerHandler(
          internal_timer_service,
          on_event_time,
          on_processing_time,
          keyed_state_backend._namespace_coder_impl,
          has_side_output,
      )
      process_timer_func = timer_handler.process_timer
  
      return (
          open_func,
          close_func,
          process_element_func,
          process_timer_func,
          internal_timer_service,
      )
  
imports: import abc
  from typing import cast
  from pyflink.common import Row
  from pyflink.common.serializer import VoidNamespaceSerializer
  from pyflink.datastream import TimeDomain, RuntimeContext
  from pyflink.datastream.functions import BroadcastProcessFunction
  from pyflink.datastream.window import WindowOperationDescriptor
  from pyflink.fn_execution import pickle
  from pyflink.fn_execution.datastream.process.input_handler import RunnerInputHandler, TimerHandler, _emit_results
  from pyflink.fn_execution.datastream import operations
  from pyflink.fn_execution.datastream.process.runtime_context import StreamingRuntimeContext
  from pyflink.fn_execution.datastream.process.process_function import InternalKeyedProcessFunctionOnTimerContext, InternalKeyedProcessFunctionContext, InternalProcessFunctionContext, InternalBroadcastProcessFunctionContext, InternalBroadcastProcessFunctionReadOnlyContext, InternalKeyedBroadcastProcessFunctionContext, InternalKeyedBroadcastProcessFunctionReadOnlyContext, InternalKeyedBroadcastProcessFunctionOnTimerContext
  from pyflink.fn_execution.datastream.process.timerservice_impl import TimerServiceImpl, InternalTimerServiceImpl, NonKeyedTimerServiceImpl
  from pyflink.fn_execution.datastream.window.window_operator import WindowOperator
  from pyflink.fn_execution.metrics.process.metric_impl import GenericMetricGroup
label_desc: Use image node_id 2 for calling a global function with example usage: extract_stateful_function(user_defined_function_proto, runtime_context, keyed_state_backend, operator_state_store) and returns: open_func,  close_func,  process_element_func,  process_timer_func,  internal_timer_service, Row, normal_data, normal_data, _on_timer, _on_timer, process_function, process_function, _on_timer, _on_timer, user_defined_func, user_defined_func, user_defined_func, window_operator, window_operator, window_operator, process_function, process_function