public CompletableFuture<Void> waitAsync(CompletableFuture<Void> eventPubFuture,
                                             String bundle,
                                             SplitDecision decision,
                                             long timeout,
                                             TimeUnit timeoutUnit) {
        return eventPubFuture
                .thenCompose(__ -> inFlightSplitRequests.computeIfAbsent(bundle, ignore -> {
                    log.info("Published the bundle split event for bundle:{}. "
                                    + "Waiting the split event to complete. Timeout: {} {}",
                            bundle, timeout, timeoutUnit);
                    CompletableFuture<Void> future = new CompletableFuture<>();
                    future.orTimeout(timeout, timeoutUnit).whenComplete((v, ex) -> {
                        if (ex != null) {
                            inFlightSplitRequests.remove(bundle);
                            log.warn("Timed out while waiting for the bundle split event: {}", bundle, ex);
                        }
                    });
                    return future;
                }))
                .whenComplete((__, ex) -> {
                    if (ex != null) {
                        log.error("Failed the bundle split event for bundle:{}", bundle, ex);
                        counter.update(Failure, Unknown);
                    } else {
                        log.info("Completed the bundle split event for bundle:{}", bundle);
                        counter.update(decision);
                    }
                });
    }
@Test
    public void testTimeout() throws IllegalAccessException {
        var counter = new SplitCounter();
        SplitManager manager = new SplitManager(counter);
        var decision = new SplitDecision();
        CompletableFuture<Void> future =
                manager.waitAsync(CompletableFuture.completedFuture(null),
                        bundle, decision, 3, TimeUnit.SECONDS);
        var inFlightUnloadRequests = getinFlightUnloadRequests(manager);

        assertEquals(inFlightUnloadRequests.size(), 1);

        try {
            future.get();
            fail();
        } catch (Exception ex) {
            assertTrue(ex.getCause() instanceof TimeoutException);
        }

        assertEquals(inFlightUnloadRequests.size(), 0);
        var counterExpected = new SplitCounter();
        counterExpected.update(SplitDecision.Label.Failure, Unknown);
        assertEquals(counter.toMetrics(null).toString(),
                counterExpected.toMetrics(null).toString());
    }