@Override
    public CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout) {

        final ArrayList<TaskManagerInfo> taskManagerInfos = new ArrayList<>(taskExecutors.size());

        for (Map.Entry<ResourceID, WorkerRegistration<WorkerType>> taskExecutorEntry :
                taskExecutors.entrySet()) {
            final ResourceID resourceId = taskExecutorEntry.getKey();
            final WorkerRegistration<WorkerType> taskExecutor = taskExecutorEntry.getValue();

            taskManagerInfos.add(
                    new TaskManagerInfo(
                            resourceId,
                            taskExecutor.getTaskExecutorGateway().getAddress(),
                            taskExecutor.getDataPort(),
                            taskExecutor.getJmxPort(),
                            taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId),
                            slotManager.getNumberRegisteredSlotsOf(taskExecutor.getInstanceID()),
                            slotManager.getNumberFreeSlotsOf(taskExecutor.getInstanceID()),
                            slotManager.getRegisteredResourceOf(taskExecutor.getInstanceID()),
                            slotManager.getFreeResourceOf(taskExecutor.getInstanceID()),
                            taskExecutor.getHardwareDescription(),
                            taskExecutor.getMemoryConfiguration(),
                            blocklistHandler.isBlockedTaskManager(taskExecutor.getResourceID())));
        }

        return CompletableFuture.completedFuture(taskManagerInfos);
    }
@Test
    void testRequestTaskManagerInfo() throws Exception {
        final ResourceID taskManagerId = ResourceID.generate();
        final TaskExecutorGateway taskExecutorGateway =
                new TestingTaskExecutorGatewayBuilder()
                        .setAddress(UUID.randomUUID().toString())
                        .createTestingTaskExecutorGateway();
        rpcService.registerGateway(taskExecutorGateway.getAddress(), taskExecutorGateway);

        resourceManager =
                new ResourceManagerBuilder().withSlotManager(createSlotManager()).buildAndStart();
        final ResourceManagerGateway resourceManagerGateway =
                resourceManager.getSelfGateway(ResourceManagerGateway.class);

        registerTaskExecutor(
                resourceManagerGateway, taskManagerId, taskExecutorGateway.getAddress());

        CompletableFuture<TaskManagerInfoWithSlots> taskManagerInfoFuture =
                resourceManagerGateway.requestTaskManagerDetailsInfo(
                        taskManagerId, TestingUtils.TIMEOUT);

        TaskManagerInfoWithSlots taskManagerInfoWithSlots = taskManagerInfoFuture.get();
        TaskManagerInfo taskManagerInfo = taskManagerInfoWithSlots.getTaskManagerInfo();

        assertThat(taskManagerInfo.getResourceId()).isEqualTo(taskManagerId);
        assertThat(taskManagerInfo.getHardwareDescription()).isEqualTo(hardwareDescription);
        assertThat(taskManagerInfo.getAddress()).isEqualTo(taskExecutorGateway.getAddress());
        assertThat(taskManagerInfo.getDataPort()).isEqualTo(dataPort);
        assertThat(taskManagerInfo.getJmxPort()).isEqualTo(jmxPort);
        assertThat(taskManagerInfo.getNumberSlots()).isEqualTo(0);
        assertThat(taskManagerInfo.getNumberAvailableSlots()).isEqualTo(0);
        assertThat(taskManagerInfoWithSlots.getAllocatedSlots()).isEmpty();
    }