docs: public static boolean verifyTopology(TopologyAPI.Topology topology) {
      if (!topology.hasName() || topology.getName().isEmpty()) {
        LOG.severe("Missing topology name");
        return false;
      }
      if (topology.getName().contains(".") || topology.getName().contains("/")) {
        LOG.severe("Invalid topology name. Topology name shouldn't have . or /");
        return false;
      }
  
      // Only verify RAM map string well-formed.
      getComponentRamMapConfig(topology);
      // Verify all bolts input streams exist. First get all output streams.
      Set<String> outputStreams = new HashSet<>();
      for (TopologyAPI.Spout spout : topology.getSpoutsList()) {
        for (TopologyAPI.OutputStream stream : spout.getOutputsList()) {
          outputStreams.add(
              stream.getStream().getComponentName() + "/" + stream.getStream().getId());
        }
      }
      for (TopologyAPI.Bolt bolt : topology.getBoltsList()) {
        for (TopologyAPI.OutputStream stream : bolt.getOutputsList()) {
          outputStreams.add(
              stream.getStream().getComponentName() + "/" + stream.getStream().getId());
        }
      }
      // Match output streams with input streams.
      for (TopologyAPI.Bolt bolt : topology.getBoltsList()) {
        for (TopologyAPI.InputStream stream : bolt.getInputsList()) {
          String key = stream.getStream().getComponentName() + "/" + stream.getStream().getId();
          if (!outputStreams.contains(key)) {
            LOG.severe("Invalid input stream " + key + " existing streams are " + outputStreams);
            return false;
          }
        }
      }
      // TODO(nbhagat): Should we enforce all output stream must be consumed?
  
      return true;
    }