Exception in thread "Flink Job" java.lang.RuntimeException: org.apache.flink.client.program.ProgramInvocationException: Job failed. (JobID: acd43401b269593f27af403379682c51)
        at StreamProcessing.Benchmark.lambda$null$0(Benchmark.java:46)
        at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.client.program.ProgramInvocationException: Job failed. (JobID: acd43401b269593f27af403379682c51)
        at org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:268)
        at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:487)
        at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:480)
        at org.apache.flink.streaming.api.environment.RemoteStreamEnvironment.executeRemotely(RemoteStreamEnvironment.java:217)
        at org.apache.flink.streaming.api.environment.RemoteStreamEnvironment.execute(RemoteStreamEnvironment.java:177)
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1510)
        at StreamProcessing.Benchmark.lambda$null$0(Benchmark.java:44)
        ... 1 more
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
        at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:146)
        at org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:265)
        ... 7 more
Caused by: java.lang.NullPointerException
        at org.apache.flink.streaming.runtime.tasks.StreamTask.createRecordWriters(StreamTask.java:1175)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:212)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:190)
        at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.<init>(SourceStreamTask.java:51)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
        at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:1405)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:689)
        ... 1 more