Hi , I am submitting the job through the WebUI. But failed with the error bellowing. Also, I found that it is a known bug https://issues.apache.org/jira/browse/FLINK-20995. Server Response Message: org.apache.flink.runtime.rest.handler.RestHandlerException: Could not execute application. at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$1(JarRunHandler.java:113) 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.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) ... 1 more 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:107) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604) ... 1 more Caused by: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Job client must be a CoordinationRequestGateway. This is a bug. 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:98) at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:84) ... 4 more 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:90) at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.setJobClient(CollectResultIterator.java:101) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.lambda$executeAsync$14(StreamExecutionEnvironment.java:2193) at java.util.ArrayList.forEach(ArrayList.java:1257) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2193) at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:206) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2169) at org.apache.flink.streaming.api.datastream.DataStream.executeAndCollectWithClient(DataStream.java:1469) at org.apache.flink.streaming.api.datastream.DataStream.executeAndCollect(DataStream.java:1334) at com.microsoft.mt.example.KafkaTest2.main(KafkaTest2.java:111) 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) ... 7 more

If I remember correctly this is due to the job expecting to return some output. For example a

SELECT * FROM x

. Flink cannot stream the results anywhere. The "fix" for the issue is to throw a more appropriate error message. In this case you need to refactor the jobs to avoid this condition, or use a different mechanism to submit them.

What kind of "some output"? In my code, the job read data from Kafka, and then sink to outer storage. Whether System.out.println() in RichSinkFunction.class is considered to "return some output" ? KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers(kafkaBrokers) .setTopics(topicName) .setGroupId(consumerGroup) .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST)) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); env.enableCheckpointing(1000 * 1 * 1, CheckpointingMode.EXACTLY_ONCE); DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka-reco-Source"); stream.addSink(new RichSinkFunction<String>() { public Logger debugLogger = LogManager.getLogger(KafkaTest.class.getName()); public ObjectStoreBondColumnTableAsyncClient<UserColumnKey> osclient = null; @Override public void setRuntimeContext(RuntimeContext t) { super.setRuntimeContext(t); } @Override public RuntimeContext getRuntimeContext() { return super.getRuntimeContext(); } @Override public IterationRuntimeContext getIterationRuntimeContext() { return super.getIterationRuntimeContext(); } @Override public void open(Configuration parameters) throws Exception { osclient = OSClientUtils.buildUserTabelClient(params); System.out.println("Open connection to other storage."); super.open(parameters); } @Override public void close() throws Exception { IOUtils.closeQuietly(osclient); System.out.println("Close connection to other storage."); } @Override public void invoke(String value, Context context) throws Exception { boolean logTag = false; if(Math.random()<0.1) { logTag=true; } if( !value.contains("xxxxxxx")) return ; Map<String,Object> jsonobj = new ObjectMapper().readValue(value, Map.class); MsnFeedbackEvent msnclick = DeserializeUtils.createMsnFeedbackEvent(jsonobj); if(!FilterUtils.msnArticleClick(msnclick)) return; System.out.println(value); Map<String, Map<String, Float>> osInput = OSClientUtils.BuildOSInput(new ArrayList<MsnFeedbackEvent>() {{add(msnclick);}}); try { OSClientUtils.BatchUpdateGeneralInfos(osclient, osInput, params); } catch (Exception e) { System.out.println("BatchUpdateGeneralInfos except: " + ExceptionUtils.getFullStackTrace(e)); } } }).name("os-reco-sink"); env.execute("reco-flink-0224"); System.out.println("reco-flink-0224");