public InstanceID registerTaskManager ( TaskManagerGateway taskManagerGateway , TaskManagerLocation taskManagerLocation , HardwareDescription resources , int numberOfSlots ) { synchronized ( this . lock ) { if ( this . isShutdown ) { throw new IllegalStateException ( "InstanceManager is shut down." ) ; } Instance prior = registeredHostsByResource . get ( taskManagerLocation . getResourceID ( ) ) ; if ( prior != null ) { throw new IllegalStateException ( "Registration attempt from TaskManager at " + taskManagerLocation . addressString ( ) + ". This connection is already registered under ID " + prior . getId ( ) ) ; } boolean wasDead = this . deadHosts . remove ( taskManagerLocation . getResourceID ( ) ) ; if ( wasDead ) { LOG . info ( "Registering TaskManager at " + taskManagerLocation . addressString ( ) + " which was marked as dead earlier because of a heart-beat timeout." ) ; } InstanceID instanceID = new InstanceID ( ) ; Instance host = new Instance ( taskManagerGateway , taskManagerLocation , instanceID , resources , numberOfSlots ) ; registeredHostsById . put ( instanceID , host ) ; registeredHostsByResource . put ( taskManagerLocation . getResourceID ( ) , host ) ; totalNumberOfAliveTaskSlots += numberOfSlots ; if ( LOG . isInfoEnabled ( ) ) { LOG . info ( String . format ( "Registered TaskManager at %s (%s) as %s. " + "Current number of registered hosts is %d. " + "Current number of alive task slots is %d." , taskManagerLocation . getHostname ( ) , taskManagerGateway . getAddress ( ) , instanceID , registeredHostsById . size ( ) , totalNumberOfAliveTaskSlots ) ) ; } host . reportHeartBeat ( ) ; // notify all listeners (for example the scheduler) notifyNewInstance ( host ) ; return instanceID ; } }
Registers a task manager . Registration of a task manager makes it available to be used for the job execution .