FLINK-14850 we changed the execute() method to be basically
Unfortunately, this means that execute() now throws an ExecutionException instead of a ProgramInvocationException or JobExecutionException as before. The ExecutionException is wrapping the other exceptions that we were throwing before.
We didn't notice this in tests because most tests use Test(Stream)Environment which overrides the execute() method and so doesn't go through the PipelineExecutor logic or the normal code path of delegating to executeAsync().
We should fix this to go back to the old behaviour.