@Override
    public void reconcileExecutionDeployments(
            ResourceID taskExecutorHost,
            ExecutionDeploymentReport executionDeploymentReport,
            Map<ExecutionAttemptID, ExecutionDeploymentState> expectedDeployedExecutions) {
        final Set<ExecutionAttemptID> unknownExecutions =
                new HashSet<>(executionDeploymentReport.getExecutions());
        final Set<ExecutionAttemptID> missingExecutions = new HashSet<>();

        for (Map.Entry<ExecutionAttemptID, ExecutionDeploymentState> execution :
                expectedDeployedExecutions.entrySet()) {
            boolean deployed = unknownExecutions.remove(execution.getKey());
            if (!deployed && execution.getValue() != ExecutionDeploymentState.PENDING) {
                missingExecutions.add(execution.getKey());
            }
        }
        if (!unknownExecutions.isEmpty()) {
            handler.onUnknownDeploymentsOf(unknownExecutions, taskExecutorHost);
        }
        if (!missingExecutions.isEmpty()) {
            handler.onMissingDeploymentsOf(missingExecutions, taskExecutorHost);
        }
    }
@Test
    void testMatchingDeployments() {
        TestingExecutionDeploymentReconciliationHandler handler =
                new TestingExecutionDeploymentReconciliationHandler();

        DefaultExecutionDeploymentReconciler reconciler =
                new DefaultExecutionDeploymentReconciler(handler);

        ResourceID resourceId = generate();
        ExecutionAttemptID attemptId = createExecutionAttemptId();

        reconciler.reconcileExecutionDeployments(
                resourceId,
                new ExecutionDeploymentReport(Collections.singleton(attemptId)),
                Collections.singletonMap(attemptId, ExecutionDeploymentState.DEPLOYED));

        assertThat(handler.getMissingExecutions()).isEmpty();
        assertThat(handler.getUnknownExecutions()).isEmpty();
    }