1. Table has been created.
    2. Table has been created.
    3. org.apache.flink.table.api.TableException: Failed to execute sql
    4.     at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:699)
    5.     at org.apache.flink.table.api.internal.StatementSetImpl.execute(StatementSetImpl.java:98)
    6.     at org.apache.zeppelin.flink.Flink112Shims.executeMultipleInsertInto(Flink112Shims.java:219)
    7.     at org.apache.zeppelin.flink.FlinkSqlInterrpeter.runSqlList(FlinkSqlInterrpeter.java:181)
    8.     at org.apache.zeppelin.flink.FlinkSqlInterrpeter.internalInterpret(FlinkSqlInterrpeter.java:119)
    9.     at org.apache.zeppelin.interpreter.AbstractInterpreter.interpret(AbstractInterpreter.java:47)
    10.     at org.apache.zeppelin.interpreter.LazyOpenInterpreter.interpret(LazyOpenInterpreter.java:110)
    11.     at org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer$InterpretJob.jobRun(RemoteInterpreterServer.java:849)
    12.     at org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer$InterpretJob.jobRun(RemoteInterpreterServer.java:741)
    13.     at org.apache.zeppelin.scheduler.Job.run(Job.java:172)
    14.     at org.apache.zeppelin.scheduler.AbstractScheduler.runJob(AbstractScheduler.java:132)
    15.     at org.apache.zeppelin.scheduler.ParallelScheduler.lambda$runJobInScheduler$0(ParallelScheduler.java:46)
    16.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    17.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    18.     at java.lang.Thread.run(Thread.java:748)
    19. Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'table_visits_stat'.
    20.     at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1918)
    21.     at org.apache.flink.api.java.ScalaShellStreamEnvironment.executeAsync(ScalaShellStreamEnvironment.java:73)
    22.     at org.apache.flink.table.planner.delegation.ExecutorBase.executeAsync(ExecutorBase.java:55)
    23.     at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:681)
    24.     ... 14 more
    25. Caused by: java.lang.RuntimeException: org.apache.flink.runtime.client.JobInitializationException: Could not instantiate JobManager.
    26.     at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:316)
    27.     at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:75)
    28.     at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)
    29.     at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
    30.     at java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:457)
    31.     at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    32.     at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    33.     at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    34.     at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:172)
    35. Caused by: org.apache.flink.runtime.client.JobInitializationException: Could not instantiate JobManager.
    36.     at org.apache.flink.runtime.dispatcher.Dispatcher.lambda$createJobManagerRunner$5(Dispatcher.java:488)
    37.     at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
    38.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    39.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    40.     at java.lang.Thread.run(Thread.java:748)
    41. Caused by: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint hdfs://stanlee-stream/flink/savepoints/table_visits_stat/savepoint-3150d3-a351d40b16fc. Cannot map checkpoint/savepoint state for operator c27dcf7b54ef6bfd6cff02ca8870b681 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.
    42.     at org.apache.flink.runtime.checkpoint.Checkpoints.throwNonRestoredStateException(Checkpoints.java:226)
    43.     at org.apache.flink.runtime.checkpoint.Checkpoints.loadAndValidateCheckpoint(Checkpoints.java:190)
    44.     at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreSavepoint(CheckpointCoordinator.java:1623)
    45.     at org.apache.flink.runtime.scheduler.SchedulerBase.tryRestoreExecutionGraphFromSavepoint(SchedulerBase.java:362)
    46.     at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:292)
    47.     at org.apache.flink.runtime.scheduler.SchedulerBase.<init>(SchedulerBase.java:249)
    48.     at org.apache.flink.runtime.scheduler.DefaultScheduler.<init>(DefaultScheduler.java:133)
    49.     at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:111)
    50.     at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:342)
    51.     at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:327)
    52.     at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.createJobMasterService(DefaultJobMasterServiceFactory.java:95)
    53.     at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.createJobMasterService(DefaultJobMasterServiceFactory.java:39)
    54.     at org.apache.flink.runtime.jobmaster.JobManagerRunnerImpl.<init>(JobManagerRunnerImpl.java:163)
    55.     at org.apache.flink.runtime.dispatcher.DefaultJobManagerRunnerFactory.createJobManagerRunner(DefaultJobManagerRunnerFactory.java:86)
    56.     at org.apache.flink.runtime.dispatcher.Dispatcher.lambda$createJobManagerRunner$5(Dispatcher.java:472)
    57.     ... 4 more