[ZEPPELIN-5510] There are some logic problems, which may lead to thread leaks

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.10.0
• Fix Version/s: 0.10.1, 0.11.0
• Component/s: Interpreters
• Assignee: gongyining
• Reporter: gongyining
• Created: 05/Sep/21 00:55
• Updated: 28/Sep/21 20:19
• Resolved: 28/Sep/21 03:58

Description

public void addJob(InterpreterContext context, JobClient jobClient) {
  String paragraphId = context.getParagraphId();
  JobClient previousJobClient = this.jobs.put(paragraphId, jobClient);
  long checkInterval = Long.parseLong(properties.getProperty("zeppelin.flink.job.check_interval", "1000"));
  FlinkJobProgressPoller thread = new FlinkJobProgressPoller(flinkWebUrl, jobClient.getJobID(), context, checkInterval);
  thread.setName("JobProgressPoller-Thread-" + paragraphId);
  thread.start();
  this.jobProgressPollerMap.put(jobClient.getJobID(), thread);
  if (previousJobClient != null) {
    LOGGER.warn("There's another Job {} that is associated with paragraph {}",
            jobClient.getJobID(), paragraphId);
  }
}

There are some problems with this code.It may cause thread leak.I think it shoud be changed  to this

public void addJob(InterpreterContext context, JobClient jobClient) {
  String paragraphId = context.getParagraphId();
  JobClient previousJobClient = this.jobs.put(paragraphId, jobClient);
  if (previousJobClient != null) {
    LOGGER.warn("There's another Job {} that is associated with paragraph {}",
            jobClient.getJobID(), paragraphId);
    return;
  }
  long checkInterval = Long.parseLong(properties.getProperty("zeppelin.flink.job.check_interval", "1000"));
  FlinkJobProgressPoller thread = new FlinkJobProgressPoller(flinkWebUrl, jobClient.getJobID(), context, checkInterval);
  thread.setName("JobProgressPoller-Thread-" + paragraphId);
  thread.start();
  this.jobProgressPollerMap.put(jobClient.getJobID(), thread);
}

If previousJobClient is not null.We shouldn't start  threading again.