@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.");
  }
@ 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 ( containerId == - 1 || containerId == 0 ) { 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 ) ) ; } LOG . fine ( "Scheduler restarted topology successfully." ) ; }
Handler to restart a topology