docs: @Override
    public boolean onKill(Scheduler.KillTopologyRequest request) {
  
      // get the topology name
      String topologyName = LocalContext.topologyName(config);
      LOG.info("Command to kill topology: " + topologyName);
  
      // set the flag that the topology being killed
      isTopologyKilled = true;
  
      synchronized (processToContainer) {
        // destroy/kill the process for each container
        for (Process p : processToContainer.keySet()) {
  
          // get the container index for the process
          int index = processToContainer.get(p);
          LOG.info("Killing executor for container: " + index);
  
          // destroy the process
          p.destroy();
          LOG.info("Killed executor for container: " + index);
        }
  
        // clear the mapping between process and container ids
        processToContainer.clear();
      }
  
      return true;
    }