Flink TableAPI在FlinkUI执行executeQuery时触发JobClient类型异常

问题现象 #

使用Flink TableAPI执行executeQuery操作时，本地运行完全正常，但通过Flink UI提交相同应用时抛出如下异常：

org.apache.flink.runtime.rest.handler.RestHandlerException: Could not execute application.
at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$1(JarRunHandler.java:108)
at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:836)
at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:811)
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1609)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.util.concurrent.CompletionException: org.apache.flink.util.FlinkRuntimeException: Could not execute application.
at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1606)
... 7 common frames omitted
Caused by: org.apache.flink.util.FlinkRuntimeException: Could not execute application.
at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:88)
at org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:70)
at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$0(JarRunHandler.java:102)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
... 7 common frames omitted
Caused by: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Failed to execute sql
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:84)
... 10 common frames omitted
Caused by: org.apache.flink.table.api.TableException: Failed to execute sql
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeQueryOperation(TableEnvironmentImpl.java:812)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1225)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:730)
at com.here.platform.extensions.pipelinetemplate.util.FlinkTableApi$.executeEnricherFunction(FlinkTableApi.scala:63)
at com.here.platform.extensions.pipelinetemplate.stream2iml.EventStream2ImlViaTableApi$.$anonfun$moveDataFromStream2Iml$1(EventStream2ImlViaTableApi.scala:95)
at com.here.platform.extensions.pipelinetemplate.stream2iml.EventStream2ImlViaTableApi$.$anonfun$moveDataFromStream2Iml$1$adapted(EventStream2ImlViaTableApi.scala:90)
at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:32)
at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:29)
at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:37)
at com.here.platform.extensions.pipelinetemplate.stream2iml.EventStream2ImlViaTableApi$.moveDataFromStream2Iml(EventStream2ImlViaTableApi.scala:90)
at com.here.platform.extensions.pipelinetemplate.usecase.EventStreamToImlConversion$.runWithEventToFeaturesConverter(EventStreamToImlConversion.scala:59)
at com.here.platform.extensions.pipelinetemplate.usecase.StreamingUseCases$.eventStreamToIml(StreamingUseCases.scala:42)
at com.here.platform.extensions.validationrepair.EventStream2Iml$.runPipeline(EventStream2Iml.scala:24)
at com.here.platform.extensions.validationrepair.EventStream2Iml$.delayedEndpoint$com$here$platform$extensions$validationrepair$EventStream2Iml$1(EventStream2Iml.scala:27)
at com.here.platform.extensions.validationrepair.EventStream2Iml$delayedInit$body.apply(EventStream2Iml.scala:16)
at scala.Function0.apply$mcV$sp(Function0.scala:34)
at scala.Function0.apply$mcV$sp$(Function0.scala:34)
at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
at scala.App.$anonfun$main$1$adapted(App.scala:76)
at scala.collection.immutable.List.foreach(List.scala:388)
at scala.App.main(App.scala:76)
at scala.App.main$(App.scala:74)
at com.here.platform.extensions.validationrepair.EventStream2Iml$.main(EventStream2Iml.scala:16)
at com.here.platform.extensions.validationrepair.EventStream2Iml.main(EventStream2Iml.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
... 13 common frames omitted
Caused by: java.lang.IllegalArgumentException: Job client must be a CoordinationRequestGateway. This is a bug.
at org.apache.flink.util.Preconditions.checkArgument(Preconditions.java:138)
at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.setJobClient(CollectResultFetcher.java:89)
at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.setJobClient(CollectResultIterator.java:101)
at org.apache.flink.table.planner.connectors.CollectDynamicSink$1.setJobClient(CollectDynamicSink.java:63)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeQueryOperation(TableEnvironmentImpl.java:797)
... 41 common frames omitted

根因分析 #

问题出在executeSql的内部逻辑：它会调用execute()和collect()方法，而通过Flink UI提交应用时采用的是detached模式，这种模式下客户端无法获取到CoordinationRequestGateway，导致collect()操作失败，抛出Job client must be a CoordinationRequestGateway异常。本地运行时是交互式模式，客户端可以正常获取该网关，因此不会报错。

解决方案 #

通过为SQL查询创建临时视图，替代直接对TableResult调用collect()的方式，避免触发依赖CoordinationRequestGateway的操作。

错误示例（导致报错） #

val result = tableEnv.executeSql("SELECT * FROM source_table")
result.collect() // 此处触发collect操作，在Flink UI运行时失败

正确示例（解决问题） #

// 创建临时视图存储查询结果
tableEnv.executeSql("CREATE TEMPORARY VIEW query_result AS SELECT * FROM source_table")
// 通过视图执行后续数据流转操作，避免直接调用collect
tableEnv.executeSql("INSERT INTO sink_table SELECT * FROM query_result")

内容的提问来源于stack exchange，提问作者Divya Jain