import javax.management.StringValueExp; //导入依赖的package包/类
public void reset(boolean includeProcessors) throws Exception {
    reset();

    // and now reset all processors for this route
    if (includeProcessors) {
        MBeanServer server = getContext().getManagementStrategy().getManagementAgent().getMBeanServer();
        if (server != null) {
            // get all the processor mbeans and sort them accordingly to their index
            String prefix = getContext().getManagementStrategy().getManagementAgent().getIncludeHostName() ? "*/" : "";
            ObjectName query = ObjectName.getInstance(jmxDomain + ":context=" + prefix + getContext().getManagementName() + ",type=processors,*");
            QueryExp queryExp = Query.match(new AttributeValueExp("RouteId"), new StringValueExp(getRouteId()));
            Set<ObjectName> names = server.queryNames(query, queryExp);
            for (ObjectName name : names) {
                server.invoke(name, "reset", null, null);
            }
        }
    }
}
 

import javax.management.StringValueExp; //导入依赖的package包/类
private boolean expireAppMasterZKSession(TwillController controller, long timeout, TimeUnit timeoutUnit) {
  MBeanServer mbeanServer = MBeanRegistry.getInstance().getPlatformMBeanServer();
  QueryExp query = Query.isInstanceOf(new StringValueExp(ConnectionMXBean.class.getName()));

  Stopwatch stopwatch = new Stopwatch();

  do {
    // Find the AM session and expire it
    Set<ObjectName> connectionBeans = mbeanServer.queryNames(ObjectName.WILDCARD, query);
    for (ObjectName objectName : connectionBeans) {

      ConnectionMXBean connectionBean = MBeanServerInvocationHandler.newProxyInstance(mbeanServer, objectName,
                                                                                      ConnectionMXBean.class, false);
      for (String node : connectionBean.getEphemeralNodes()) {
        if (node.endsWith("/instances/" + controller.getRunId().getId())) {
          // This is the AM, expire the session.
          LOG.info("Kill AM session {}", connectionBean.getSessionId());
          connectionBean.terminateSession();
          return true;
        }
      }
    }
  } while (stopwatch.elapsedTime(timeoutUnit) < timeout);

  return false;
}
 

import javax.management.StringValueExp; //导入依赖的package包/类
private boolean expireAppMasterZKSession(TwillController controller, long timeout, TimeUnit timeoutUnit) {
  MBeanServer mbeanServer = MBeanRegistry.getInstance().getPlatformMBeanServer();
  QueryExp query = Query.isInstanceOf(new StringValueExp(ConnectionMXBean.class.getName()));

  Stopwatch stopwatch = new Stopwatch();
  stopwatch.start();
  do {
    // Find the AM session and expire it
    Set<ObjectName> connectionBeans = mbeanServer.queryNames(ObjectName.WILDCARD, query);
    for (ObjectName objectName : connectionBeans) {

      ConnectionMXBean connectionBean = MBeanServerInvocationHandler.newProxyInstance(mbeanServer, objectName,
                                                                                      ConnectionMXBean.class, false);
      for (String node : connectionBean.getEphemeralNodes()) {
        if (node.endsWith("/instances/" + controller.getRunId().getId())) {
          // This is the AM, expire the session.
          LOG.info("Kill AM session {}", connectionBean.getSessionId());
          connectionBean.terminateSession();
          return true;
        }
      }
    }
    Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
  } while (stopwatch.elapsedTime(timeoutUnit) < timeout);

  return false;
}