Jaehyeon Kim created FLINK-33404:
------------------------------------
Summary: on_timer method is missing in ProcessFunction and
CoProcessFunction of Pyflink
Key: FLINK-33404
URL: https://issues.apache.org/jira/browse/FLINK-33404
Project: Flink
Issue Type: Bug
Components: API / Python
Reporter: Jaehyeon Kim
Hello,
I find the `on_timer` method is not found in ProcessFunction and
CoProcessFunction of Pyflink and it causes an error when I register a timer eg)
```
...
File
"/home/jaehyeon/personal/flink-demos/venv/lib/python3.8/site-packages/pyflink/fn_execution/datastream/process/input_handler.py",
line 101, in process_timer
yield from _emit_results(
File
"/home/jaehyeon/personal/flink-demos/venv/lib/python3.8/site-packages/pyflink/fn_execution/datastream/process/input_handler.py",
line 131, in _emit_results
for result in results:
File
"/home/jaehyeon/personal/flink-demos/venv/lib/python3.8/site-packages/pyflink/fn_execution/datastream/process/input_handler.py",
line 114, in _on_processing_time
yield from self._on_processing_time_func(timestamp, key, namespace)
File
"/home/jaehyeon/personal/flink-demos/venv/lib/python3.8/site-packages/pyflink/fn_execution/datastream/process/operations.py",
line 308, in on_processing_time
return _on_timer(TimeDomain.PROCESSING_TIME, timestamp, key)
File
"/home/jaehyeon/personal/flink-demos/venv/lib/python3.8/site-packages/pyflink/fn_execution/datastream/process/operations.py",
line 317, in _on_timer
return process_function.on_timer(timestamp, on_timer_ctx)
AttributeError: 'ReadingFilter' object has no attribute 'on_timer'
at
org.apache.beam.runners.fnexecution.control.FnApiControlClient$ResponseStreamObserver.onNext(FnApiControlClient.java:180)
at
org.apache.beam.runners.fnexecution.control.FnApiControlClient$ResponseStreamObserver.onNext(FnApiControlClient.java:160)
at
org.apache.beam.vendor.grpc.v1p48p1.io.grpc.stub.ServerCalls$StreamingServerCallHandler$StreamingServerCallListener.onMessage(ServerCalls.java:262)
at
org.apache.beam.vendor.grpc.v1p48p1.io.grpc.ForwardingServerCallListener.onMessage(ForwardingServerCallListener.java:33)
at
org.apache.beam.vendor.grpc.v1p48p1.io.grpc.Contexts$ContextualizedServerCallListener.onMessage(Contexts.java:76)
at
org.apache.beam.vendor.grpc.v1p48p1.io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messagesAvailableInternal(ServerCallImpl.java:332)
at
org.apache.beam.vendor.grpc.v1p48p1.io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messagesAvailable(ServerCallImpl.java:315)
at
org.apache.beam.vendor.grpc.v1p48p1.io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1MessagesAvailable.runInContext(ServerImpl.java:834)
at
org.apache.beam.vendor.grpc.v1p48p1.io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
at
org.apache.beam.vendor.grpc.v1p48p1.io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
... 3 more
```
I'm working on Pyflink 1.17.1 but it would be applicable other versions.
Can the method be added to the functions?
Cheers,
Jaehyeon
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
