code: public void submitTopology(StormTopology topology, boolean isLocal) throws Exception
      {
          if (isLocal)
          {
              // ローカル環境で実行
              LocalCluster cluster = new LocalCluster();
              cluster.submitTopology(this.topologyName, this.config, topology);
          }
          else
          {
              // 分散環境で実行
              try
              {
                  StormSubmitter.submitTopology(this.topologyName, this.config, topology);
              }
              catch (Exception ex)
              {
                  String logFormat = "Occur exception at Topology Submit. Skip Topology Submit. : TopologyName={0}";
                  logger.error(MessageFormat.format(logFormat, this.topologyName), ex);
              }
          }
      }
docs: Topologyを実行する。
  
  @param topology StormTopology
  @param isLocal true:ローカルモード、false:分散モード
  @throws Exception Topology実行失敗時