WeiZhong94 commented on a change in pull request #15149:
URL: https://github.com/apache/flink/pull/15149#discussion_r595669961
##########
File path: flink-python/pyflink/fn_execution/operation_utils.py
##########
@@ -293,52 +283,122 @@ def wrapped_process_function(value):
def extract_keyed_process_function(user_defined_function_proto, ctx,
on_timer_ctx,
collector, keyed_state_backend):
+ func_type = user_defined_function_proto.function_type
+ UserDefinedDataStreamFunction =
flink_fn_execution_pb2.UserDefinedDataStreamFunction
+ func = None
process_function = pickle.loads(user_defined_function_proto.payload)
- process_element = process_function.process_element
on_timer = process_function.on_timer
- def wrapped_keyed_process_function(value):
- if value[0] is not None:
- # it is timer data
- # VALUE: TIMER_FLAG, TIMESTAMP_OF_TIMER, CURRENT_WATERMARK,
CURRENT_KEY_OF_TIMER, None
- on_timer_ctx.set_timestamp(value[1])
- on_timer_ctx.timer_service().set_current_watermark(value[2])
- current_key = value[3]
- on_timer_ctx.set_current_key(current_key)
- keyed_state_backend.set_current_key(current_key)
- if value[0] ==
KeyedProcessFunctionInputFlag.EVENT_TIME_TIMER.value:
- on_timer_ctx.set_time_domain(TimeDomain.EVENT_TIME)
- elif value[0] ==
KeyedProcessFunctionInputFlag.PROC_TIME_TIMER.value:
- on_timer_ctx.set_time_domain(TimeDomain.PROCESSING_TIME)
+ if func_type == UserDefinedDataStreamFunction.KEYED_PROCESS:
+ process_element = process_function.process_element
+
+ def wrapped_keyed_process_function(value):
+ if value[0] is not None:
+ # it is timer data
+ # VALUE:
+ # TIMER_FLAG, TIMESTAMP_OF_TIMER, CURRENT_WATERMARK,
CURRENT_KEY_OF_TIMER, None
+ on_timer_ctx.set_timestamp(value[1])
+ on_timer_ctx.timer_service().set_current_watermark(value[2])
+ state_current_key = value[3]
+ user_current_key = state_current_key[0]
+ on_timer_ctx.set_current_key(user_current_key)
+ keyed_state_backend.set_current_key(state_current_key)
+ if value[0] ==
KeyedProcessFunctionInputFlag.EVENT_TIME_TIMER.value:
Review comment:
The semantic of `TimerOperandType` is different from
`KeyedProcessFunctionInputFlag`. Do you mean to use `TimeOperandType` instead
of `KeyedProcessFunctionOutputFlag`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
