private void handleDecision(
            @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
                    Optional<Decision> decisionOpt) {
        Decision decision =
                decisionOpt.orElseGet(
                        () -> callWithLock(() -> spillStrategy.decideActionWithGlobalInfo(this)));

        if (!decision.getBufferToSpill().isEmpty()) {
            spillBuffers(decision.getBufferToSpill());
        }
        if (!decision.getBufferToRelease().isEmpty()) {
            releaseBuffers(decision.getBufferToRelease());
        }
    }
@Test
    void testHandleDecision() throws Exception {
        final int targetSubpartition = 0;
        final int numFinishedBufferToTriggerDecision = 4;
        List<BufferIndexAndChannel> toSpill =
                HybridShuffleTestUtils.createBufferIndexAndChannelsList(
                        targetSubpartition, 0, 1, 2);
        List<BufferIndexAndChannel> toRelease =
                HybridShuffleTestUtils.createBufferIndexAndChannelsList(targetSubpartition, 2, 3);
        HsSpillingStrategy spillingStrategy =
                TestingSpillingStrategy.builder()
                        .setOnBufferFinishedFunction(
                                (numFinishedBuffers, poolSize) -> {
                                    if (numFinishedBuffers < numFinishedBufferToTriggerDecision) {
                                        return Optional.of(Decision.NO_ACTION);
                                    }
                                    return Optional.of(
                                            Decision.builder()
                                                    .addBufferToSpill(targetSubpartition, toSpill)
                                                    .addBufferToRelease(
                                                            targetSubpartition, toRelease)
                                                    .build());
                                })
                        .build();
        CompletableFuture<List<SpilledBuffer>> spilledFuture = new CompletableFuture<>();
        CompletableFuture<Integer> readableFuture = new CompletableFuture<>();
        TestingFileDataIndex dataIndex =
                TestingFileDataIndex.builder()
                        .setAddBuffersConsumer(spilledFuture::complete)
                        .setMarkBufferReadableConsumer(
                                (subpartitionId, bufferIndex) ->
                                        readableFuture.complete(bufferIndex))
                        .build();
        HsMemoryDataManager memoryDataManager =
                createMemoryDataManager(spillingStrategy, dataIndex);
        for (int i = 0; i < 4; i++) {
            memoryDataManager.append(
                    createRecord(i), targetSubpartition, Buffer.DataType.DATA_BUFFER);
        }

        assertThatFuture(spilledFuture).eventuallySucceeds();
        assertThatFuture(readableFuture).eventuallySucceeds();
        assertThat(readableFuture).isCompletedWithValue(2);
        assertThat(memoryDataManager.getNumTotalUnSpillBuffers()).isEqualTo(1);
    }