You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Dian Fu (Jira)" <ji...@apache.org> on 2022/04/22 14:41:00 UTC
[jira] [Closed] (FLINK-22984) UnsupportedOperationException when using Python UDF to generate watermark
[ https://issues.apache.org/jira/browse/FLINK-22984?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Dian Fu closed FLINK-22984.
---------------------------
Fix Version/s: 1.16.0
1.13.7
1.14.5
1.15.1
Assignee: Juntao Hu
Resolution: Fixed
Fixed in:
- master via 7ce5a7c6e1eab6823094a94bc0bca30d0ee618f1
- release-1.15 via 703b10ca5d004e8e79059e814fcf8503f84e2da8
- release-1.14 via 0806ad5a154e37d09b53ce56d59cec8dc11209da
- release-1.13 via 79a86f35fb321cb5f8dd40442db8c6bafb00153c
> UnsupportedOperationException when using Python UDF to generate watermark
> -------------------------------------------------------------------------
>
> Key: FLINK-22984
> URL: https://issues.apache.org/jira/browse/FLINK-22984
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 1.13.0, 1.13.1
> Reporter: Maciej Bryński
> Assignee: Juntao Hu
> Priority: Minor
> Labels: auto-deprioritized-critical, auto-deprioritized-major, pull-request-available
> Fix For: 1.16.0, 1.13.7, 1.14.5, 1.15.1
>
>
> Hi,
> I'm trying to use output of Python UDF (parse_data) to set watermark for the table
> {code:java}
> CREATE TABLE test (
> data BYTES,
> ts as parse_data(data).ts,
> WATERMARK for ts as ts
> ) WITH (
> 'connector' = 'kafka',
> 'topic' = 'test',
> 'properties.bootstrap.servers' = 'localhost:9092',
> 'properties.group.id' = 'flink',
> 'scan.startup.mode' = 'earliest-offset',
> 'format' = 'raw'
> ){code}
> Then running SELECT on this table gives me exception
> {code:java}
> Py4JJavaError: An error occurred while calling o311.hasNext.
> : java.lang.RuntimeException: Failed to fetch next result
> at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:109)
> at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:80)
> at org.apache.flink.table.api.internal.TableResultImpl$CloseableRowIteratorWrapper.hasNext(TableResultImpl.java:370)
> at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.base/java.lang.reflect.Method.invoke(Method.java:566)
> at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
> at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
> at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
> at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
> at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
> at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
> at java.base/java.lang.Thread.run(Thread.java:829)
> Caused by: java.io.IOException: Failed to fetch job execution result
> at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:177)
> at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.next(CollectResultFetcher.java:120)
> at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:106)
> ... 13 more
> Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
> at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2022)
> at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:175)
> ... 15 more
> Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
> at org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:137)
> at java.base/java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:680)
> at java.base/java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:658)
> at java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2094)
> at org.apache.flink.runtime.minicluster.MiniClusterJobClient.getJobExecutionResult(MiniClusterJobClient.java:134)
> at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:174)
> ... 15 more
> Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
> at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138)
> at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82)
> at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:207)
> at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:197)
> at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:188)
> at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:677)
> at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79)
> at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:435)
> at jdk.internal.reflect.GeneratedMethodAccessor31.invoke(Unknown Source)
> at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.base/java.lang.reflect.Method.invoke(Method.java:566)
> at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305)
> at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212)
> at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77)
> at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158)
> at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
> at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
> at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
> at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
> at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
> at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
> at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
> at akka.actor.Actor$class.aroundReceive(Actor.scala:517)
> at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
> at akka.actor.ActorCell.invoke(ActorCell.scala:561)
> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
> at akka.dispatch.Mailbox.run(Mailbox.scala:225)
> at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
> at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> Caused by: java.lang.RuntimeException: Generated WatermarkGenerator fails to generate for row: +I([2, 10, ..., 23]).
> at org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec$DefaultWatermarkGeneratorSupplier$DefaultWatermarkGenerator.onEvent(WatermarkPushDownSpec.java:172)
> at org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec$DefaultWatermarkGeneratorSupplier$DefaultWatermarkGenerator.onEvent(WatermarkPushDownSpec.java:150)
> at org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionStateWithWatermarkGenerator.onEvent(KafkaTopicPartitionStateWithWatermarkGenerator.java:82)
> at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordsWithTimestamps(AbstractFetcher.java:368)
> at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:183)
> at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.runFetchLoop(KafkaFetcher.java:142)
> at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:826)
> at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
> at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:66)
> at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:269)
> Caused by: java.lang.UnsupportedOperationException: This method is a placeholder and should not be called.
> at org.apache.flink.table.functions.python.PythonScalarFunction.eval(PythonScalarFunction.java:69)
> at WatermarkGenerator$14.currentWatermark(Unknown Source)
> at org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec$DefaultWatermarkGeneratorSupplier$DefaultWatermarkGenerator.onEvent(WatermarkPushDownSpec.java:166)
> ... 9 more
> {code}
--
This message was sent by Atlassian Jira
(v8.20.7#820007)