public void init(int numThreads, boolean abortOnCorruption) throws IOException {
    // We have numThreads executor + one timer thread used for timing out
    // procedures and triggering periodic procedures.
    this.corePoolSize = numThreads;
    this.maxPoolSize = 10 * numThreads;
    LOG.info("Starting {} core workers (bigger of cpus/4 or 16) with max (burst) worker count={}",
      corePoolSize, maxPoolSize);

    this.threadGroup = new ThreadGroup("PEWorkerGroup");
    this.timeoutExecutor = new TimeoutExecutorThread<>(this, threadGroup, "ProcExecTimeout");
    this.workerMonitorExecutor = new TimeoutExecutorThread<>(this, threadGroup, "WorkerMonitor");
    ThreadFactory backingThreadFactory = new ThreadFactory() {

      @Override
      public Thread newThread(Runnable r) {
        return new Thread(threadGroup, r);
      }
    };
    int size = Math.max(2, Runtime.getRuntime().availableProcessors());
    ThreadPoolExecutor executor =
      new ThreadPoolExecutor(size, size, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(),
        new ThreadFactoryBuilder().setDaemon(true)
          .setNameFormat(getClass().getSimpleName() + "-Async-Task-Executor-%d")
          .setThreadFactory(backingThreadFactory).build());
    executor.allowCoreThreadTimeOut(true);
    this.asyncTaskExecutor = executor;
    forceUpdateExecutor = Executors.newFixedThreadPool(1, new ThreadFactoryBuilder().setDaemon(true)
      .setNameFormat("Force-Update-PEWorker-%d").setThreadFactory(backingThreadFactory).build());
    store.registerListener(new ProcedureStoreListener() {

      @Override
      public void forceUpdate(long[] procIds) {
        Arrays.stream(procIds).forEach(procId -> forceUpdateExecutor.execute(() -> {
          try {
            forceUpdateProcedure(procId);
          } catch (IOException e) {
            LOG.warn("Failed to force update procedure with pid={}", procId);
          }
        }));
      }
    });

    // Create the workers
    workerId.set(0);
    workerThreads = new CopyOnWriteArrayList<>();
    for (int i = 0; i < corePoolSize; ++i) {
      workerThreads.add(new WorkerThread(threadGroup));
    }

    long st, et;

    // Acquire the store lease.
    st = System.nanoTime();
    store.recoverLease();
    et = System.nanoTime();
    LOG.info("Recovered {} lease in {}", store.getClass().getSimpleName(),
      StringUtils.humanTimeDiff(TimeUnit.NANOSECONDS.toMillis(et - st)));

    // start the procedure scheduler
    scheduler.start();

    // TODO: Split in two steps.
    // TODO: Handle corrupted procedures (currently just a warn)
    // The first one will make sure that we have the latest id,
    // so we can start the threads and accept new procedures.
    // The second step will do the actual load of old procedures.
    st = System.nanoTime();
    load(abortOnCorruption);
    et = System.nanoTime();
    LOG.info("Loaded {} in {}", store.getClass().getSimpleName(),
      StringUtils.humanTimeDiff(TimeUnit.NANOSECONDS.toMillis(et - st)));
  }
Initialize the procedure executor, but do not start workers. We will start them later.
<p/>
It calls ProcedureStore.recoverLease() and ProcedureStore.load() to recover the lease, and
ensure a single executor, and start the procedure replay to resume and recover the previous
pending and in-progress procedures.
@param numThreads        number of threads available for procedure execution.
@param abortOnCorruption true if you want to abort your service in case a corrupted procedure
                         is found on replay. otherwise false.
hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java