flink/flink1.15-shims/src/main/java/org/apache/zeppelin/flink/Flink115SqlInterpreter.java [182:210]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  public Flink115SqlInterpreter(FlinkSqlContext flinkSqlContext, boolean isBatch) {
    this.flinkSqlContext = flinkSqlContext;
    this.isBatch = isBatch;
    if (isBatch) {
      this.tbenv = (TableEnvironment) flinkSqlContext.getBtenv();
    } else {
      this.tbenv = (TableEnvironment) flinkSqlContext.getStenv();
    }
    this.z = (ZeppelinContext) flinkSqlContext.getZeppelinContext();
    this.sqlParser = ((TableEnvironmentInternal) tbenv).getParser();
    this.sqlSplitter = new SqlSplitter();
    JobListener jobListener = new JobListener() {
      @Override
      public void onJobSubmitted(@Nullable JobClient jobClient, @Nullable Throwable throwable) {
        if (lock.isHeldByCurrentThread()) {
          lock.unlock();
          LOGGER.info("UnLock JobSubmitLock");
        }
      }

      @Override
      public void onJobExecuted(@Nullable JobExecutionResult jobExecutionResult, @Nullable Throwable throwable) {

      }
    };

    ((ExecutionEnvironment) flinkSqlContext.getBenv()).registerJobListener(jobListener);
    ((StreamExecutionEnvironment) flinkSqlContext.getSenv()).registerJobListener(jobListener);
  }
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/Flink116SqlInterpreter.java [182:210]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  public Flink116SqlInterpreter(FlinkSqlContext flinkSqlContext, boolean isBatch) {
    this.flinkSqlContext = flinkSqlContext;
    this.isBatch = isBatch;
    if (isBatch) {
      this.tbenv = (TableEnvironment) flinkSqlContext.getBtenv();
    } else {
      this.tbenv = (TableEnvironment) flinkSqlContext.getStenv();
    }
    this.z = (ZeppelinContext) flinkSqlContext.getZeppelinContext();
    this.sqlParser = ((TableEnvironmentInternal) tbenv).getParser();
    this.sqlSplitter = new SqlSplitter();
    JobListener jobListener = new JobListener() {
      @Override
      public void onJobSubmitted(@Nullable JobClient jobClient, @Nullable Throwable throwable) {
        if (lock.isHeldByCurrentThread()) {
          lock.unlock();
          LOGGER.info("UnLock JobSubmitLock");
        }
      }

      @Override
      public void onJobExecuted(@Nullable JobExecutionResult jobExecutionResult, @Nullable Throwable throwable) {

      }
    };

    ((ExecutionEnvironment) flinkSqlContext.getBenv()).registerJobListener(jobListener);
    ((StreamExecutionEnvironment) flinkSqlContext.getSenv()).registerJobListener(jobListener);
  }
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



flink/flink1.17-shims/src/main/java/org/apache/zeppelin/flink/Flink117SqlInterpreter.java [182:210]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  public Flink117SqlInterpreter(FlinkSqlContext flinkSqlContext, boolean isBatch) {
    this.flinkSqlContext = flinkSqlContext;
    this.isBatch = isBatch;
    if (isBatch) {
      this.tbenv = (TableEnvironment) flinkSqlContext.getBtenv();
    } else {
      this.tbenv = (TableEnvironment) flinkSqlContext.getStenv();
    }
    this.z = (ZeppelinContext) flinkSqlContext.getZeppelinContext();
    this.sqlParser = ((TableEnvironmentInternal) tbenv).getParser();
    this.sqlSplitter = new SqlSplitter();
    JobListener jobListener = new JobListener() {
      @Override
      public void onJobSubmitted(@Nullable JobClient jobClient, @Nullable Throwable throwable) {
        if (lock.isHeldByCurrentThread()) {
          lock.unlock();
          LOGGER.info("UnLock JobSubmitLock");
        }
      }

      @Override
      public void onJobExecuted(@Nullable JobExecutionResult jobExecutionResult, @Nullable Throwable throwable) {

      }
    };

    ((ExecutionEnvironment) flinkSqlContext.getBenv()).registerJobListener(jobListener);
    ((StreamExecutionEnvironment) flinkSqlContext.getSenv()).registerJobListener(jobListener);
  }
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -