docs: @VisibleForTesting
    void restartTopologyHandler(String topologyName) throws TopologyRuntimeManagementException {
      assert !potentialStaleExecutionData;
      Integer containerId = Context.topologyContainerId(config);
      Scheduler.RestartTopologyRequest restartTopologyRequest =
          Scheduler.RestartTopologyRequest.newBuilder()
              .setTopologyName(topologyName)
              .setContainerIndex(containerId)
              .build();
      // If we restart the container including TMaster, wee need to clean TMasterLocation,
      // since when starting up, TMaster expects no other existing TMaster,
      // i.e. TMasterLocation does not exist
      if (containerId == -1 || containerId == 0) {
        // get the instance of state manager to clean state
        SchedulerStateManagerAdaptor stateManager = Runtime.schedulerStateManagerAdaptor(runtime);
        Boolean result = stateManager.deleteTMasterLocation(topologyName);
        if (result == null || !result) {
          throw new TopologyRuntimeManagementException(
              "Failed to clear TMaster location. Check whether TMaster set it correctly.");
        }
      }
  
      if (!schedulerClient.restartTopology(restartTopologyRequest)) {
        throw new TopologyRuntimeManagementException(String.format(
            "Failed to restart topology '%s'", topologyName));
      }
      // Clean the connection when we are done.
      LOG.fine("Scheduler restarted topology successfully.");
    }