Uploaded image for project: 'Flink'
  1. Flink
  2. FLINK-22984

UnsupportedOperationException when using Python UDF to generate watermark

    XMLWordPrintableJSON

    Details

      Description

      Hi,

      I'm trying to use output of Python UDF (parse_data) to set watermark for the table

      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'
      )

      Then running SELECT on this table gives me exception

      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
      

        Attachments

          Activity

            People

            • Assignee:
              Unassigned
              Reporter:
              maver1ck Maciej BryƄski
            • Votes:
              0 Vote for this issue
              Watchers:
              4 Start watching this issue

              Dates

              • Created:
                Updated: