Register TaskManager to this pool, only those slots come from registered TaskManager will be considered valid.
Also it provides a way for us to keep "dead" or "abnormal" TaskManagers out of this pool.

@param resourceID The id of the TaskManager
18429
@Override
	public boolean registerTaskManager(final ResourceID resourceID) {

		componentMainThreadExecutor.assertRunningInMainThread();

		log.debug("Register new TaskExecutor {}.", resourceID);
		return registeredTaskManagers.add(resourceID);
	}
96474
protected function registerResource(ResourceInterface $resource): void
    {
        \assert($this->hasNotBeenAdded($resource));

        $this->addedResources[$resource->getId()][$resource->getType()] = true;
    }
1238206
@Override
  public void register(TaskExecutor taskExecutor) {
	lock.lock();
	try {
	  Iterator<TaskExecutor> it = taskExecutors.iterator();
	  while (it.hasNext()) {
		TaskExecutor e = it.next();
		if (e.isShutdown() && e.hasTask() == false) {
		  it.remove();
		}
	  }

	  if (taskExecutors.contains(taskExecutor) == false) {
		taskExecutors.add(taskExecutor);
	  }
	} finally {
	  lock.unlock();
	}
  }
317711
@Override
    @SuppressWarnings("deprecation")
    public void registerOperations(ManagementResourceRegistration resourceRegistration) {
        if (addHandler != null) {
            registerAddOperation(resourceRegistration, addHandler, addRestartLevel);
        }
        if (removeHandler != null) {
            registerRemoveOperation(resourceRegistration, removeHandler, removeRestartLevel);
        }
    }
857514
public void unregisterTaskManager(InstanceID instanceId, boolean terminated){
		Instance instance = registeredHostsById.get(instanceId);

		if (instance != null){
			registeredHostsById.remove(instance.getId());
			registeredHostsByResource.remove(instance.getTaskManagerID());

			if (terminated) {
				deadHosts.add(instance.getTaskManagerID());
			}

			instance.markDead();

			totalNumberOfAliveTaskSlots -= instance.getTotalNumberOfSlots();

			notifyDeadInstance(instance);

			LOG.info(
				"Unregistered task manager " + instance.getTaskManagerLocation().addressString() +
				". Number of registered task managers " + getNumberOfRegisteredTaskManagers() +
				". Number of available slots " + getTotalNumberOfSlots() + ".");
		} else {
			LOG.warn("Tried to unregister instance {} but it is not registered.", instanceId);
		}
	}
930843
public static function register(string $filterId = self::FILTER_IDENTIFIER): void
    {
        if (!empty(self::$filterId)) {
            throw new RuntimeException('Stream filter already registered');
        }

        $result = stream_filter_register($filterId, __CLASS__);
        if ($result === false) {
            throw new RuntimeException('Stream filter was not registered');
        }
        self::$filterId = $filterId;
    }
198376
@Override
  public void onContainerStopped(final ContainerId containerId) {
    final boolean hasContainer = this.containers.hasContainer(containerId.toString());
    if (hasContainer) {
      this.reefEventHandlers.onResourceStatus(
          ResourceStatusEventImpl.newBuilder()
             .setIdentifier(containerId.toString())
             .setState(State.DONE)
             .build());
    }
  }
44071
private void unregisterTask(final ExecutionVertexID id) {

		// Task de-registration must be atomic
		synchronized (this) {

			final Task task = this.runningTasks.remove(id);
			if (task == null) {
				LOG.error("Cannot find task with ID " + id + " to unregister");
				return;
			}

			// remove the local tmp file for unregistered tasks.
			for (Entry<String, String> e: DistributedCache.getCachedFile(task.getEnvironment().getJobConfiguration())) {
				this.fileCache.deleteTmpFile(e.getKey(), task.getJobID());
			}
			// Unregister task from the byte buffered channel manager
			this.channelManager.unregister(id, task);

			// Unregister task from profiling
			task.unregisterProfiler(this.profiler);

			// Unregister task from memory manager
			task.unregisterMemoryManager(this.memoryManager);

			// Unregister task from library cache manager
			try {
				LibraryCacheManager.unregister(task.getJobID());
			} catch (IOException e) {
				if (LOG.isDebugEnabled()) {
					LOG.debug("Unregistering the job vertex ID " + id + " caused an IOException");
				}
			}
		}
	}
971670
public void registerWorkManager(WorkManager wm)
   {
      if (wm != null)
      {
         if (wm.getName() == null || wm.getName().trim().equals(""))
            throw new IllegalArgumentException("The name of WorkManager is invalid: " + wm);

         if (trace)
            log.tracef("Registering WorkManager: %s", wm);

         if (!workmanagers.keySet().contains(wm.getName()))
         {
            workmanagers.put(wm.getName(), wm);

            // Replay events for distributed work managers
            if (wm instanceof DistributedWorkManager)
            {
               WorkManagerEventQueue wmeq = WorkManagerEventQueue.getInstance();
               List<WorkManagerEvent> events = wmeq.getEvents(wm.getName());

               if (!events.isEmpty())
               {
                  if (trace)
                     log.tracef("%s: Events=%s", wm.getName(), events);

                  for (WorkManagerEvent event : events)
                  {
                     if (event.getType() == WorkManagerEvent.TYPE_JOIN)
                     {
                        DistributedWorkManager dwm = resolveDistributedWorkManager(event.getAddress());

                        if (dwm != null)
                        {
                           Collection<NotificationListener> copy =
                              new ArrayList<>(dwm.getNotificationListeners());
                           for (NotificationListener nl : copy)
                           {
                              nl.join(event.getAddress());
                           }
                        }
                     }
                     else if (event.getType() == WorkManagerEvent.TYPE_LEAVE)
                     {
                        DistributedWorkManager dwm = 
                           (DistributedWorkManager)activeWorkmanagers.get(event.getAddress().getWorkManagerId());

                        if (dwm != null)
                        {
                           Collection<NotificationListener> copy =
                              new ArrayList<NotificationListener>(dwm.getNotificationListeners());
                           for (NotificationListener nl : copy)
                           {
                              nl.leave(event.getAddress());
                           }
                        }

                        removeWorkManager(event.getAddress().getWorkManagerId());
                     }
                     else if (event.getType() == WorkManagerEvent.TYPE_UPDATE_SHORT_RUNNING)
                     {
                        DistributedWorkManager dwm = 
                           (DistributedWorkManager)activeWorkmanagers.get(event.getAddress().getWorkManagerId());

                        if (dwm != null)
                        {
                           Collection<NotificationListener> copy =
                              new ArrayList<NotificationListener>(dwm.getNotificationListeners());
                           for (NotificationListener nl : copy)
                           {
                              nl.updateShortRunningFree(event.getAddress(), event.getValue());
                           }
                        }
                     }
                     else if (event.getType() == WorkManagerEvent.TYPE_UPDATE_LONG_RUNNING)
                     {
                        DistributedWorkManager dwm = 
                           (DistributedWorkManager)activeWorkmanagers.get(event.getAddress().getWorkManagerId());

                        if (dwm != null)
                        {
                           Collection<NotificationListener> copy =
                              new ArrayList<NotificationListener>(dwm.getNotificationListeners());
                           for (NotificationListener nl : copy)
                           {
                              nl.updateLongRunningFree(event.getAddress(), event.getValue());
                           }
                        }
                     }
                  }
               }
            }
         }
      }
   }
150060
protected boolean scheduleNewTask(String taskId) {
    // Put it to the pending start queue
    LOG.info(String.format("We are to schedule task: [%s]", taskId));

    // Update the tasksId
    int containerIndex = TaskUtils.getContainerIndexForTaskId(taskId);
    tasksId.put(containerIndex, taskId);

    // Re-schedule it
    toScheduleTasks.add(taskId);
    LOG.info(String.format("Added task: %s into the to-schedule-tasks queue: ", taskId));
    return true;
  }
1251114
public void put(String ID,
                    String resourceName,
                    int dir,
                    boolean visible) {
        registerEntry(ID, new ResourceEntry(resourceName, dir), visible);
    }
1196209
public static String getResourceInstitution(String resourceId)
	{
		Resource resource = getResource(resourceId);
    	return resource.getDataInstitution();
	}
1168721
public void setResourceContainerImageId(java.util.Collection<StringFilter> resourceContainerImageId) {
        if (resourceContainerImageId == null) {
            this.resourceContainerImageId = null;
            return;
        }

        this.resourceContainerImageId = new java.util.ArrayList<StringFilter>(resourceContainerImageId);
    }
744671
public void unload(String resourceId) {
        LoadableResource res = this.resources.get(resourceId);
        if (Objects.nonNull(res)) {
            res.unload();
        }
    }
753388
public SlotReport createSlotReport(ResourceID resourceId) {
		final int numberSlots = taskSlots.size();

		List<SlotStatus> slotStatuses = Arrays.asList(new SlotStatus[numberSlots]);

		for (int i = 0; i < numberSlots; i++) {
			TaskSlot taskSlot = taskSlots.get(i);
			SlotID slotId = new SlotID(resourceId, taskSlot.getIndex());

			SlotStatus slotStatus = new SlotStatus(
				slotId,
				taskSlot.getResourceProfile(),
				taskSlot.getJobId(),
				taskSlot.getAllocationId());

			slotStatuses.set(i, slotStatus);
		}

		final SlotReport slotReport = new SlotReport(slotStatuses);

		return slotReport;
	}
773558
public synchronized void setFinished(final int taskId) {
    final TaskEntity task = getTask(taskId, runningTasks);
    runningTasks.remove(task);
    finishedTasks.add(task);
  }
606502
@Override
    public void write(String resourceId, byte[] data) {
        try {
            File file = this.cachedResources.get(resourceId);
            if (Objects.isNull(file)) {
                file = new File(localDir, resourceId + SUFFIX);
                Files.write(file.toPath(), data);
                this.cachedResources.put(resourceId, file);
            } else {
            	Files.write(file.toPath(), data);
            }
        } catch (Exception e) {
            LOG.log(Level.WARNING, "Caching of resource failed: " + resourceId, e);
        }
    }