docs: @VisibleForTesting
    void killTopologyHandler(String topologyName) throws TopologyRuntimeManagementException {
      Scheduler.KillTopologyRequest killTopologyRequest = Scheduler.KillTopologyRequest.newBuilder()
          .setTopologyName(topologyName).build();
  
      if (!schedulerClient.killTopology(killTopologyRequest)) {
        throw new TopologyRuntimeManagementException(
            String.format("Failed to kill topology '%s' with scheduler, "
                + "please re-try the kill command later", topologyName));
      }
  
      // clean up the state of the topology in state manager
      cleanState(topologyName, Runtime.schedulerStateManagerAdaptor(runtime));
  
      if (potentialStaleExecutionData) {
        LOG.warning(String.format("Topology %s does not exist. Cleaned up potential stale state.",
            topologyName));
      } else {
        LOG.fine(String.format("Scheduler killed topology %s successfully.", topologyName));
      }
    }