[
https://issues.apache.org/jira/browse/FLINK-33404?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-33404:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Not a Priority
> Labels: pull-request-available
>
> 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)
