public void close() {
        inFlightSplitRequests.forEach((bundle, future) -> {
            if (!future.isDone()) {
                String msg = String.format("Splitting bundle: %s, but the manager already closed.", bundle);
                log.warn(msg);
                future.completeExceptionally(new IllegalStateException(msg));
            }
        });
        inFlightSplitRequests.clear();
    }
@Test
    public void testClose() throws IllegalAccessException {
        SplitManager manager = new SplitManager(new SplitCounter());
        var decision = new SplitDecision();
        CompletableFuture<Void> future =
                manager.waitAsync(CompletableFuture.completedFuture(null),
                        bundle, decision, 5, TimeUnit.SECONDS);
        var inFlightUnloadRequests = getinFlightUnloadRequests(manager);
        assertEquals(inFlightUnloadRequests.size(), 1);
        manager.close();
        assertEquals(inFlightUnloadRequests.size(), 0);

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