@Override
    public void connectToResourceManager(ResourceManagerGateway resourceManagerGateway) {
        assertHasBeenStarted();

        resourceRequirementServiceConnectionManager.connect(
                resourceRequirements ->
                        resourceManagerGateway.declareRequiredResources(
                                jobMasterId, resourceRequirements, rpcTimeout));

        declareResourceRequirements(declarativeSlotPool.getResourceRequirements());
    }
@Test
    void testConnectToResourceManagerDeclaresRequiredResources() throws Exception {
        final Collection<ResourceRequirement> requiredResources =
                Arrays.asList(
                        ResourceRequirement.create(ResourceProfile.UNKNOWN, 2),
                        ResourceRequirement.create(ResourceProfile.ZERO, 4));

        try (DeclarativeSlotPoolService declarativeSlotPoolService =
                createDeclarativeSlotPoolService(
                        new TestingDeclarativeSlotPoolFactory(
                                new TestingDeclarativeSlotPoolBuilder()
                                        .setGetResourceRequirementsSupplier(
                                                () -> requiredResources)))) {
            final TestingResourceManagerGateway resourceManagerGateway =
                    new TestingResourceManagerGateway();

            final CompletableFuture<ResourceRequirements> declaredResourceRequirements =
                    new CompletableFuture<>();

            resourceManagerGateway.setDeclareRequiredResourcesFunction(
                    (jobMasterId, resourceRequirements) -> {
                        declaredResourceRequirements.complete(resourceRequirements);
                        return CompletableFuture.completedFuture(Acknowledge.get());
                    });

            declarativeSlotPoolService.connectToResourceManager(resourceManagerGateway);

            final ResourceRequirements resourceRequirements = declaredResourceRequirements.join();

            assertThat(resourceRequirements.getResourceRequirements()).isEqualTo(requiredResources);
            assertThat(resourceRequirements.getJobId()).isEqualTo(jobId);
            assertThat(resourceRequirements.getTargetAddress()).isEqualTo(address);
        }
    }