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 ; }
Verify if the given topology has all the necessary information