@SuppressWarnings("FieldAccessNotGuarded")
    // Note that: callWithLock ensure that code block guarded by resultPartitionReadLock and
    // subpartitionLock.
    public List<BufferWithIdentity> spillSubpartitionBuffers(
            List<BufferIndexAndChannel> toSpill, CompletableFuture<Void> spillDoneFuture) {
        return callWithLock(
                () ->
                        toSpill.stream()
                                .map(
                                        indexAndChannel -> {
                                            int bufferIndex = indexAndChannel.getBufferIndex();
                                            return startSpillingBuffer(bufferIndex, spillDoneFuture)
                                                    .map(
                                                            (context) ->
                                                                    new BufferWithIdentity(
                                                                            context.getBuffer(),
                                                                            bufferIndex,
                                                                            targetChannel));
                                        })
                                .filter(Optional::isPresent)
                                .map(Optional::get)
                                .collect(Collectors.toList()));
    }
@Test
    void testSpillSubpartitionBuffers() throws Exception {
        CompletableFuture<Void> spilledDoneFuture = new CompletableFuture<>();
        TestingMemoryDataManagerOperation memoryDataManagerOperation =
                TestingMemoryDataManagerOperation.builder()
                        .setRequestBufferFromPoolSupplier(() -> createBufferBuilder(RECORD_SIZE))
                        .build();
        HsSubpartitionMemoryDataManager subpartitionMemoryDataManager =
                createSubpartitionMemoryDataManager(memoryDataManagerOperation);
        final int numBuffers = 3;
        for (int i = 0; i < numBuffers; i++) {
            subpartitionMemoryDataManager.append(createRecord(i), DataType.DATA_BUFFER);
        }

        List<BufferIndexAndChannel> toStartSpilling =
                HybridShuffleTestUtils.createBufferIndexAndChannelsList(0, 0, 1, 2);
        List<BufferWithIdentity> buffers =
                subpartitionMemoryDataManager.spillSubpartitionBuffers(
                        toStartSpilling, spilledDoneFuture);
        assertThat(toStartSpilling)
                .zipSatisfy(
                        buffers,
                        (expected, spilled) -> {
                            assertThat(expected.getBufferIndex())
                                    .isEqualTo(spilled.getBufferIndex());
                            assertThat(expected.getChannel()).isEqualTo(spilled.getChannelIndex());
                        });
        List<Integer> expectedValues = Arrays.asList(0, 1, 2);
        checkBuffersRefCountAndValue(buffers, Arrays.asList(2, 2, 2), expectedValues);
        spilledDoneFuture.complete(null);
        checkBuffersRefCountAndValue(buffers, Arrays.asList(1, 1, 1), expectedValues);
    }