@Override
    public void onBufferFinished() {
        Optional<Decision> decision =
                spillStrategy.onBufferFinished(numUnSpillBuffers.incrementAndGet(), getPoolSize());
        handleDecision(decision);
    }
@Test
    void testHandleEmptyDecision() throws Exception {
        CompletableFuture<Void> globalDecisionFuture = new CompletableFuture<>();
        HsSpillingStrategy spillingStrategy =
                TestingSpillingStrategy.builder()
                        .setOnBufferFinishedFunction(
                                (finishedBuffer, poolSize) -> {
                                    // return empty optional to trigger global decision.
                                    return Optional.empty();
                                })
                        .setDecideActionWithGlobalInfoFunction(
                                (provider) -> {
                                    globalDecisionFuture.complete(null);
                                    return Decision.NO_ACTION;
                                })
                        .build();
        HsMemoryDataManager memoryDataManager = createMemoryDataManager(spillingStrategy);
        // trigger an empty decision.
        memoryDataManager.onBufferFinished();
        assertThat(globalDecisionFuture).isCompleted();
    }