@Test public void testCreateWorkerSlotProfiles() { final Configuration config = new Configuration(); config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "100m"); config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 5); final ResourceProfile rmCalculatedResourceProfile = ResourceManager.createWorkerSlotProfiles(config).iterator().next(); final ResourceProfile tmCalculatedResourceProfile = TaskManagerServices.computeSlotResourceProfile( config.getInteger(TaskManagerOptions.NUM_TASK_SLOTS), MemorySize.parse(config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE)).getBytes()); assertEquals(rmCalculatedResourceProfile, tmCalculatedResourceProfile); }
public static Collection<ResourceProfile> createWorkerSlotProfiles(Configuration config) { final int numSlots = config.getInteger(TaskManagerOptions.NUM_TASK_SLOTS); final long managedMemoryBytes = MemorySize.parse(config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE)).getBytes(); final ResourceProfile resourceProfile = TaskManagerServices.computeSlotResourceProfile(numSlots, managedMemoryBytes); return Collections.nCopies(numSlots, resourceProfile); }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { public static Collection<ResourceProfile> createWorkerSlotProfiles(Configuration config) { final int numSlots = config.getInteger(TaskManagerOptions.NUM_TASK_SLOTS); final long managedMemoryBytes = MemorySize.parse(config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE)).getBytes(); final ResourceProfile resourceProfile = TaskManagerServices.computeSlotResourceProfile(numSlots, managedMemoryBytes); return Collections.nCopies(numSlots, resourceProfile); } }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { public static Collection<ResourceProfile> createWorkerSlotProfiles(Configuration config) { final int numSlots = config.getInteger(TaskManagerOptions.NUM_TASK_SLOTS); final long managedMemoryBytes = MemorySize.parse(config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE)).getBytes(); final ResourceProfile resourceProfile = TaskManagerServices.computeSlotResourceProfile(numSlots, managedMemoryBytes); return Collections.nCopies(numSlots, resourceProfile); } ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup); }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { public static Collection<ResourceProfile> createWorkerSlotProfiles(Configuration config) { final int numSlots = config.getInteger(TaskManagerOptions.NUM_TASK_SLOTS); final long managedMemoryBytes = MemorySize.parse(config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE)).getBytes(); final ResourceProfile resourceProfile = TaskManagerServices.computeSlotResourceProfile(numSlots, managedMemoryBytes); return Collections.nCopies(numSlots, resourceProfile); } ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<RegistrationResponse> registerJobManager(
			final JobMasterId jobMasterId,
			final ResourceID jobManagerResourceId,
			final String jobManagerAddress,
			final JobID jobId,
			final Time timeout); @Override CompletableFuture<RegistrationResponse> registerTaskExecutor(
			final String taskExecutorAddress,
			final ResourceID taskExecutorResourceId,
			final int dataPort,
			final HardwareDescription hardwareDescription,
			final Time timeout); @Override CompletableFuture<Acknowledge> sendSlotReport(ResourceID taskManagerResourceId, InstanceID taskManagerRegistrationId, SlotReport slotReport, Time timeout); @Override void heartbeatFromTaskManager(final ResourceID resourceID, final SlotReport slotReport); @Override void heartbeatFromJobManager(final ResourceID resourceID); @Override void disconnectTaskManager(final ResourceID resourceId, final Exception cause); @Override void disconnectJobManager(final JobID jobId, final Exception cause); @Override CompletableFuture<Acknowledge> requestSlot(
			JobMasterId jobMasterId,
			SlotRequest slotRequest,
			final Time timeout); @Override void cancelSlotRequest(AllocationID allocationID); @Override void notifySlotAvailable(
			final InstanceID instanceID,
			final SlotID slotId,
			final AllocationID allocationId); @Override CompletableFuture<Acknowledge> deregisterApplication(
			final ApplicationStatus finalStatus,
			@Nullable final String diagnostics); @Override CompletableFuture<Integer> getNumberOfRegisteredTaskManagers(); @Override CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout); @Override CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(ResourceID resourceId, Time timeout); @Override CompletableFuture<ResourceOverview> requestResourceOverview(Time timeout); @Override CompletableFuture<Collection<Tuple2<ResourceID, String>>> requestTaskManagerMetricQueryServiceAddresses(Time timeout); @Override CompletableFuture<TransientBlobKey> requestTaskManagerFileUpload(ResourceID taskManagerId, FileType fileType, Time timeout); @Override void grantLeadership(final UUID newLeaderSessionID); @Override void revokeLeadership(); @Override void handleError(final Exception exception); @VisibleForTesting abstract Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); abstract boolean stopWorker(WorkerType worker); static Collection<ResourceProfile> createWorkerSlotProfiles(Configuration config); }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { public static Collection<ResourceProfile> createWorkerSlotProfiles(Configuration config) { final int numSlots = config.getInteger(TaskManagerOptions.NUM_TASK_SLOTS); final long managedMemoryBytes = MemorySize.parse(config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE)).getBytes(); final ResourceProfile resourceProfile = TaskManagerServices.computeSlotResourceProfile(numSlots, managedMemoryBytes); return Collections.nCopies(numSlots, resourceProfile); } ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<RegistrationResponse> registerJobManager(
			final JobMasterId jobMasterId,
			final ResourceID jobManagerResourceId,
			final String jobManagerAddress,
			final JobID jobId,
			final Time timeout); @Override CompletableFuture<RegistrationResponse> registerTaskExecutor(
			final String taskExecutorAddress,
			final ResourceID taskExecutorResourceId,
			final int dataPort,
			final HardwareDescription hardwareDescription,
			final Time timeout); @Override CompletableFuture<Acknowledge> sendSlotReport(ResourceID taskManagerResourceId, InstanceID taskManagerRegistrationId, SlotReport slotReport, Time timeout); @Override void heartbeatFromTaskManager(final ResourceID resourceID, final SlotReport slotReport); @Override void heartbeatFromJobManager(final ResourceID resourceID); @Override void disconnectTaskManager(final ResourceID resourceId, final Exception cause); @Override void disconnectJobManager(final JobID jobId, final Exception cause); @Override CompletableFuture<Acknowledge> requestSlot(
			JobMasterId jobMasterId,
			SlotRequest slotRequest,
			final Time timeout); @Override void cancelSlotRequest(AllocationID allocationID); @Override void notifySlotAvailable(
			final InstanceID instanceID,
			final SlotID slotId,
			final AllocationID allocationId); @Override CompletableFuture<Acknowledge> deregisterApplication(
			final ApplicationStatus finalStatus,
			@Nullable final String diagnostics); @Override CompletableFuture<Integer> getNumberOfRegisteredTaskManagers(); @Override CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout); @Override CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(ResourceID resourceId, Time timeout); @Override CompletableFuture<ResourceOverview> requestResourceOverview(Time timeout); @Override CompletableFuture<Collection<Tuple2<ResourceID, String>>> requestTaskManagerMetricQueryServiceAddresses(Time timeout); @Override CompletableFuture<TransientBlobKey> requestTaskManagerFileUpload(ResourceID taskManagerId, FileType fileType, Time timeout); @Override void grantLeadership(final UUID newLeaderSessionID); @Override void revokeLeadership(); @Override void handleError(final Exception exception); @VisibleForTesting abstract Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); abstract boolean stopWorker(WorkerType worker); static Collection<ResourceProfile> createWorkerSlotProfiles(Configuration config); static final String RESOURCE_MANAGER_NAME; }