public static CompletableFuture<Map<String, String>> labelFailure(
            final Throwable cause,
            final Context context,
            final Executor mainThreadExecutor,
            final Collection<FailureEnricher> failureEnrichers) {
        // list of CompletableFutures to enrich failure with labels from each enricher
        final Collection<CompletableFuture<Map<String, String>>> enrichFutures = new ArrayList<>();

        for (final FailureEnricher enricher : failureEnrichers) {
            enrichFutures.add(
                    enricher.processFailure(cause, context)
                            .thenApply(
                                    enricherLabels -> {
                                        final Map<String, String> validLabels = new HashMap<>();
                                        enricherLabels.forEach(
                                                (k, v) -> {
                                                    if (!enricher.getOutputKeys().contains(k)) {
                                                        LOG.warn(
                                                                "Ignoring label with key {} from enricher {}"
                                                                        + " violating contract, keys allowed {}.",
                                                                k,
                                                                enricher.getClass(),
                                                                enricher.getOutputKeys());
                                                    } else {
                                                        validLabels.put(k, v);
                                                    }
                                                });
                                        return validLabels;
                                    })
                            .exceptionally(
                                    t -> {
                                        LOG.warn(
                                                "Enricher {} threw an exception.",
                                                enricher.getClass(),
                                                t);
                                        return Collections.emptyMap();
                                    }));
        }
        // combine all CompletableFutures into a single CompletableFuture containing a Map of labels
        return FutureUtils.combineAll(enrichFutures)
                .thenApplyAsync(
                        labelsToMerge -> {
                            final Map<String, String> mergedLabels = new HashMap<>();
                            for (Map<String, String> labels : labelsToMerge) {
                                labels.forEach(
                                        (k, v) ->
                                                // merge label with existing, throwing an exception
                                                // if there is a key conflict
                                                mergedLabels.merge(
                                                        k,
                                                        v,
                                                        (first, second) -> {
                                                            throw new FlinkRuntimeException(
                                                                    String.format(
                                                                            MERGE_EXCEPTION_MSG,
                                                                            k));
                                                        }));
                            }
                            return mergedLabels;
                        },
                        mainThreadExecutor);
    }
@Test
    public void testLabelFutureWithValidEnricher() {
        // validate labelFailure by enricher with correct outputKeys
        final Throwable cause = new RuntimeException("test exception");
        final Set<FailureEnricher> failureEnrichers = new HashSet<>();
        final FailureEnricher validEnricher = new TestEnricher("enricherKey");
        failureEnrichers.add(validEnricher);

        final CompletableFuture<Map<String, String>> result =
                FailureEnricherUtils.labelFailure(
                        cause,
                        null,
                        ComponentMainThreadExecutorServiceAdapter.forMainThread(),
                        failureEnrichers);

        assertThatFuture(result)
                .eventuallySucceeds()
                .satisfies(
                        labels -> {
                            assertThat(labels).hasSize(1);
                            assertThat(labels).containsKey("enricherKey");
                            assertThat(labels).containsValue("enricherKeyValue");
                        });
    }