public void close() {
        inFlightUnloadRequest.forEach((bundle, future) -> {
            if (!future.isDone()) {
                String msg = String.format("Unloading bundle: %s, but the unload manager already closed.", bundle);
                log.warn(msg);
                future.completeExceptionally(new IllegalStateException(msg));
            }
        });
        inFlightUnloadRequest.clear();
    }
@Test
    public void testClose() 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,5, TimeUnit.SECONDS);
        Map<String, CompletableFuture<Void>> inFlightUnloadRequestMap = getInFlightUnloadRequestMap(manager);
        assertEquals(inFlightUnloadRequestMap.size(), 1);
        manager.close();
        assertEquals(inFlightUnloadRequestMap.size(), 0);

        try {
            future.get();
            fail();
        } catch (Exception ex) {
            assertTrue(ex.getCause() instanceof IllegalStateException);
        }
        assertEquals(counter.getBreakdownCounters().get(Failure).get(Unknown).get(), 1);
    }