code: @Override
  	public void newInstanceAvailable(Instance instance) {
  		if (instance == null) {
  			throw new IllegalArgumentException();
  		}
  		if (instance.getNumberOfAvailableSlots() <= 0) {
  			throw new IllegalArgumentException("The given instance has no resources.");
  		}
  		if (!instance.isAlive()) {
  			throw new IllegalArgumentException("The instance is not alive.");
  		}
  		
  		// synchronize globally for instance changes
  		synchronized (this.globalLock) {
  			
  			// check we do not already use this instance
  			if (!this.allInstances.add(instance)) {
  				throw new IllegalArgumentException("The instance is already contained.");
  			}
  			
  			try {
  				// make sure we get notifications about slots becoming available
  				instance.setSlotAvailabilityListener(this);
  				
  				// store the instance in the by-host-lookup
  				String instanceHostName = instance.getTaskManagerLocation().getHostname();
  				Set<Instance> instanceSet = allInstancesByHost.get(instanceHostName);
  				if (instanceSet == null) {
  					instanceSet = new HashSet<Instance>();
  					allInstancesByHost.put(instanceHostName, instanceSet);
  				}
  				instanceSet.add(instance);
  
  				// add it to the available resources and let potential waiters know
  				this.instancesWithAvailableResources.put(instance.getTaskManagerID(), instance);
  
  				// add all slots as available
  				for (int i = 0; i < instance.getNumberOfAvailableSlots(); i++) {
  					newSlotAvailable(instance);
  				}
  			}
  			catch (Throwable t) {
  				LOG.error("Scheduler could not add new instance " + instance, t);
  				removeInstance(instance);
  			}
  		}
  	}
docs: --------------------------------------------------------------------------------------------