Hello Don t we need `on timer ` method in the `CoProcessFunc Apache Flink #troubleshooting

Hello, Don't we need `on_timer()` method in the `...

Jaehyeon Kim

10/27/2023, 7:04 AM

Hello, Don't we need

on_timer()

method in the

CoProcessFunction

of Pyflink? I have an error because of that. Can you inform me where I can report this?

Copy code

...
  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

Jaehyeon Kim

10/28/2023, 5:43 AM

Just had a bit more search about the process function family.

on_timer

method not exists • ProcessFunction • CoProcessFunction function not exists • ProcessJoinFunction match to java source • KeyedProcessFunction • BroadcastProcessFunction • KeyedBroadcastProcessFunction I requested a JIRA account and plan to create a ticket about them.

Jaehyeon Kim

10/31/2023, 3:14 AM

I created two tickets. on_timer method is missing in ProcessFunction and CoProcessFunction of Pyflink https://issues.apache.org/jira/browse/FLINK-33404 ProcessJoinFunction not found in Pyflink https://issues.apache.org/jira/browse/FLINK-33405

Open in Slack

Previous Next