[Thread-leakage]DruidTaskResolver does not stop poll thread after tranquilizer is closed

[realAaronWu]

Every time a new connection is established in IndexService, there is a DruidTaskResolver submitting a runnable to a new thread (poll thread) bind to the connection and periodically check the task running status, but when the beam is closed, the poll runnable cannot be notified and cancled properly. This caused a thread leakage in a long-last running jvm.

[realAaronWu]

The single-threaded executor does not return a future, so the runnable cannot be stopped when the beam is closed.

pollExecutor.submit(
        new Runnable
        {
          override def run(): Unit = {
            while (!Thread.currentThread().isInterrupted) {
              try {
                val tasksSnapshot: Map[Long, (String, AtomicReference[Addr], Updatable[Addr])] = lock.synchronized {
                  while (timekeeper.now.millis < nextPoll) {
                    val waitMillis = nextPoll - timekeeper.now.millis
                    if (waitMillis > 0) {
                      lock.wait(waitMillis)
                    }
                  }
                  tasks
                }

                val now = timekeeper.now
                val baseWait = now.plus(pollPeriod).millis - now.millis
                val fuzzyWait = (math.max(1 + 0.25 * Random.nextGaussian, 0) * baseWait).toLong
                nextPoll = now.millis + fuzzyWait

                val running = Await.result(indexService.runningTasks())

                log.debug(s"Polled[$indexServiceKey] and found ${running.size} tasks " +
                  s"(${running.count(_._2.isBound)} bound).")

                for ((taskNumber, (taskId, addr, updatable)) <- tasksSnapshot) {
                  val runningHostPort = running.get(taskId)
                  val runningAddr: Addr = runningHostPort.map(_.toAddr).getOrElse(Addr.Neg)
                  if (addr.get() != runningAddr) {
                    log.info(s"Updating location for task[$taskId] (#$taskNumber) to[$runningAddr].")
                    addr.set(runningAddr)
                    updatable.update(runningAddr)
                  }
                }
              }
              catch {
                case NonFatal(e) =>
                  log.warn(e, s"Poll failed, trying again at[${new DateTime(nextPoll)}].")
              }
            }
          }
        }
      )