code: public static String get_topology_id(StormClusterState zkCluster, String storm_name) throws Exception {
          List<String> active_storms = zkCluster.active_storms();
          String rtn = null;
          if (active_storms != null) {
              for (String topology_id : active_storms) {
  
                  if (!topology_id.contains(storm_name)) {
                      continue;
                  }
                  StormBase base = zkCluster.storm_base(topology_id, null);
                  if (base != null && storm_name.equals(Common.getTopologyNameById(topology_id))) {
                      rtn = topology_id;
                      break;
                  }
              }
          }
          return rtn;
      }
docs: if a topology's name is equal to the input storm_name, then return the topology id, otherwise return null