code: public void checkNewWorkers(Map conf) throws IOException, InterruptedException {
          Set<String> workers = new HashSet<>();
          for (Entry<String, Pair<Integer, Integer>> entry : workerIdToStartTimeAndPort.entrySet()) {
              String workerId = entry.getKey();
              int startTime = entry.getValue().getFirst();
              LocalState ls = StormConfig.worker_state(conf, workerId);
              WorkerHeartbeat whb = (WorkerHeartbeat) ls.get(Common.LS_WORKER_HEARTBEAT);
              if (whb == null) {
                  if ((TimeUtils.current_time_secs() - startTime) <
                          JStormUtils.parseInt(conf.get(Config.SUPERVISOR_WORKER_START_TIMEOUT_SECS))) {
                      LOG.info(workerId + " still hasn't started");
                  } else {
                      LOG.error("Failed to start Worker " + workerId);
                      workers.add(workerId);
                  }
              } else {
                  LOG.info("Successfully started worker " + workerId);
                  workers.add(workerId);
              }
          }
          for (String workerId : workers) {
              Integer port = this.workerIdToStartTimeAndPort.get(workerId).getSecond();
              this.workerIdToStartTimeAndPort.remove(workerId);
              this.portToWorkerId.remove(port);
          }
      }
docs: check whether timestamp of new workers is not > SUPERVISOR_WORKER_START_TIMEOUT_SECS, otherwise mark as failed