@Override
    public Decision decideActionWithGlobalInfo(HsSpillingInfoProvider spillingInfoProvider) {
        Decision.Builder builder = Decision.builder();
        // Save the cost of lock, if pool size is changed between checkSpill and checkRelease, pool
        // size checker will handle this inconsistency.
        int poolSize = spillingInfoProvider.getPoolSize();
        checkSpill(spillingInfoProvider, poolSize, builder);
        checkRelease(spillingInfoProvider, poolSize, builder);
        return builder.build();
    }
@Test
    void testDecideActionWithGlobalInfo() {
        final int subpartition1 = 0;
        final int subpartition2 = 1;

        List<BufferIndexAndChannel> subpartitionBuffers1 =
                createBufferIndexAndChannelsList(subpartition1, 1, 2, 3, 4, 5);
        List<BufferIndexAndChannel> subpartitionBuffers2 =
                createBufferIndexAndChannelsList(subpartition2, 1, 2, 3, 4, 5);

        TestingSpillingInfoProvider spillInfoProvider =
                TestingSpillingInfoProvider.builder()
                        .setGetNumSubpartitionsSupplier(() -> NUM_SUBPARTITIONS)
                        .addSubpartitionBuffers(subpartition1, subpartitionBuffers1)
                        .addSubpartitionBuffers(subpartition2, subpartitionBuffers2)
                        .addSpillBuffers(subpartition1, Arrays.asList(0, 1, 2, 3))
                        .addSpillBuffers(subpartition2, Arrays.asList(1, 2, 3))
                        .setGetNumTotalUnSpillBuffersSupplier(
                                () -> (int) (10 * NUM_BUFFERS_TRIGGER_SPILLING_RATIO))
                        .setGetNumTotalRequestedBuffersSupplier(() -> 10)
                        .setGetPoolSizeSupplier(() -> 10)
                        .build();

        Decision decision = spillStrategy.decideActionWithGlobalInfo(spillInfoProvider);

        // all not spilled buffers need to spill.
        Map<Integer, List<BufferIndexAndChannel>> expectedSpillBuffers = new HashMap<>();
        expectedSpillBuffers.put(subpartition1, subpartitionBuffers1.subList(4, 5));
        expectedSpillBuffers.put(
                subpartition2, new ArrayList<>(subpartitionBuffers2.subList(0, 1)));
        expectedSpillBuffers.get(subpartition2).addAll(subpartitionBuffers2.subList(4, 5));
        assertThat(decision.getBufferToSpill()).isEqualTo(expectedSpillBuffers);

        Map<Integer, List<BufferIndexAndChannel>> expectedReleaseBuffers = new HashMap<>();
        expectedReleaseBuffers.put(
                subpartition1, new ArrayList<>(subpartitionBuffers1.subList(0, 3)));
        expectedReleaseBuffers.put(
                subpartition2, new ArrayList<>(subpartitionBuffers2.subList(1, 4)));
        assertThat(decision.getBufferToRelease()).isEqualTo(expectedReleaseBuffers);
    }