docs: private RegistrationResponse registerTaskExecutorInternal(
  			TaskExecutorGateway taskExecutorGateway,
  			String taskExecutorAddress,
  			ResourceID taskExecutorResourceId,
  			int dataPort,
  			HardwareDescription hardwareDescription) {
  		WorkerRegistration<WorkerType> oldRegistration = taskExecutors.remove(taskExecutorResourceId);
  		if (oldRegistration != null) {
  			// TODO :: suggest old taskExecutor to stop itself
  			log.debug("Replacing old registration of TaskExecutor {}.", taskExecutorResourceId);
  
  			// remove old task manager registration from slot manager
  			slotManager.unregisterTaskManager(oldRegistration.getInstanceID());
  		}
  
  		final WorkerType newWorker = workerStarted(taskExecutorResourceId);
  
  		if (newWorker == null) {
  			log.warn("Discard registration from TaskExecutor {} at ({}) because the framework did " +
  				"not recognize it", taskExecutorResourceId, taskExecutorAddress);
  			return new RegistrationResponse.Decline("unrecognized TaskExecutor");
  		} else {
  			WorkerRegistration<WorkerType> registration =
  				new WorkerRegistration<>(taskExecutorGateway, newWorker, dataPort, hardwareDescription);
  
  			log.info("Registering TaskManager with ResourceID {} ({}) at ResourceManager", taskExecutorResourceId, taskExecutorAddress);
  			taskExecutors.put(taskExecutorResourceId, registration);
  
  			taskManagerHeartbeatManager.monitorTarget(taskExecutorResourceId, new HeartbeatTarget<Void>() {
  				@Override
  				public void receiveHeartbeat(ResourceID resourceID, Void payload) {
  					// the ResourceManager will always send heartbeat requests to the
  					// TaskManager
  				}
  
  				@Override
  				public void requestHeartbeat(ResourceID resourceID, Void payload) {
  					taskExecutorGateway.heartbeatFromResourceManager(resourceID);
  				}
  			});
  
  			return new TaskExecutorRegistrationSuccess(
  				registration.getInstanceID(),
  				resourceId,
  				clusterInformation);
  		}
  	}