public CompletableFuture<Void> waitAsync(CompletableFuture<Void> eventPubFuture,
                                             String bundle,
                                             UnloadDecision decision,
                                             long timeout,
                                             TimeUnit timeoutUnit) {
        return eventPubFuture.thenCompose(__ -> inFlightUnloadRequest.computeIfAbsent(bundle, ignore -> {
            if (log.isDebugEnabled()) {
                log.debug("Handle unload bundle: {}, timeout: {} {}", bundle, timeout, timeoutUnit);
            }
            CompletableFuture<Void> future = new CompletableFuture<>();
            future.orTimeout(timeout, timeoutUnit).whenComplete((v, ex) -> {
                if (ex != null) {
                    inFlightUnloadRequest.remove(bundle);
                    log.warn("Failed to wait unload for serviceUnit: {}", bundle, ex);
                }
            });
            return future;
        })).whenComplete((__, ex) -> {
            if (ex != null) {
                counter.update(Failure, Unknown);
                log.warn("Failed to unload bundle: {}", bundle, ex);
                return;
            }
            log.info("Complete unload bundle: {}", bundle);
            counter.update(decision);
        });
    }
@Test
    public void testTimeout() throws IllegalAccessException {
        UnloadCounter counter = new UnloadCounter();
        UnloadManager manager = new UnloadManager(counter, "mockBrokerId");
        var unloadDecision =
                new UnloadDecision(new Unload("broker-1", "bundle-1"), Success, Admin);
        CompletableFuture<Void> future =
                manager.waitAsync(CompletableFuture.completedFuture(null),
                        "bundle-1", unloadDecision, 3, TimeUnit.SECONDS);
        Map<String, CompletableFuture<Void>> inFlightUnloadRequestMap = getInFlightUnloadRequestMap(manager);

        assertEquals(inFlightUnloadRequestMap.size(), 1);

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

        assertEquals(inFlightUnloadRequestMap.size(), 0);
        assertEquals(counter.getBreakdownCounters().get(Failure).get(Unknown).get(), 1);
    }