public void submitTopology ( ) throws TopologySubmissionException { // build primary runtime config first Config primaryRuntime = Config . newBuilder ( ) . putAll ( LauncherUtils . getInstance ( ) . createPrimaryRuntime ( topology ) ) . build ( ) ; // call launcher directly here if in dry-run mode if ( Context . dryRun ( config ) ) { callLauncherRunner ( primaryRuntime ) ; return ; } // 1. Do prepare work // create an instance of state manager String statemgrClass = Context . stateManagerClass ( config ) ; IStateManager statemgr ; // Create an instance of the launcher class String launcherClass = Context . launcherClass ( config ) ; ILauncher launcher ; // create an instance of the uploader class String uploaderClass = Context . uploaderClass ( config ) ; IUploader uploader ; // create an instance of state manager try { statemgr = ReflectionUtils . newInstance ( statemgrClass ) ; } catch ( IllegalAccessException | InstantiationException | ClassNotFoundException e ) { throw new TopologySubmissionException ( String . format ( "Failed to instantiate state manager class '%s'" , statemgrClass ) , e ) ; } // create an instance of launcher try { launcher = ReflectionUtils . newInstance ( launcherClass ) ; } catch ( IllegalAccessException | InstantiationException | ClassNotFoundException e ) { throw new LauncherException ( String . format ( "Failed to instantiate launcher class '%s'" , launcherClass ) , e ) ; } // create an instance of uploader try { uploader = ReflectionUtils . newInstance ( uploaderClass ) ; } catch ( IllegalAccessException | InstantiationException | ClassNotFoundException e ) { throw new UploaderException ( String . format ( "Failed to instantiate uploader class '%s'" , uploaderClass ) , e ) ; } // Put it in a try block so that we can always clean resources try { // initialize the state manager statemgr . initialize ( config ) ; // initialize the uploader uploader . initialize ( config ) ; // TODO(mfu): timeout should read from config SchedulerStateManagerAdaptor adaptor = new SchedulerStateManagerAdaptor ( statemgr , 5000 ) ; // Check if topology is already running validateSubmit ( adaptor , topology . getName ( ) ) ; LOG . log ( Level . FINE , "Topology {0} to be submitted" , topology . getName ( ) ) ; Config runtimeWithoutPackageURI = Config . newBuilder ( ) . putAll ( primaryRuntime ) . putAll ( LauncherUtils . getInstance ( ) . createAdaptorRuntime ( adaptor ) ) . put ( Key . LAUNCHER_CLASS_INSTANCE , launcher ) . build ( ) ; PackingPlan packingPlan = LauncherUtils . getInstance ( ) . createPackingPlan ( config , runtimeWithoutPackageURI ) ; // The packing plan might call for a number of containers different than the config // settings. If that's the case we need to modify the configs to match. runtimeWithoutPackageURI = updateNumContainersIfNeeded ( runtimeWithoutPackageURI , topology , packingPlan ) ; // If the packing plan is valid we will upload necessary packages URI packageURI = uploadPackage ( uploader ) ; // Update the runtime config with the packageURI Config runtimeAll = Config . newBuilder ( ) . putAll ( runtimeWithoutPackageURI ) . put ( Key . TOPOLOGY_PACKAGE_URI , packageURI ) . build ( ) ; callLauncherRunner ( runtimeAll ) ; } catch ( LauncherException | PackingException e ) { // we undo uploading of topology package only if launcher fails to // launch topology, which will throw LauncherException or PackingException uploader . undo ( ) ; throw e ; } finally { SysUtils . closeIgnoringExceptions ( uploader ) ; SysUtils . closeIgnoringExceptions ( launcher ) ; SysUtils . closeIgnoringExceptions ( statemgr ) ; } }
Submit a topology 1 . Instantiate necessary resources 2 . Valid whether it is legal to submit a topology 3 . Call LauncherRunner