@SuppressWarnings("FieldAccessNotGuarded")
    // Note that: callWithLock ensure that code block guarded by resultPartitionReadLock and
    // subpartitionLock.
    public Deque<BufferIndexAndChannel> getBuffersSatisfyStatus(
            SpillStatus spillStatus, ConsumeStatusWithId consumeStatusWithId) {
        return callWithLock(
                () -> {
                    // TODO return iterator to avoid completely traversing the queue for each call.
                    Deque<BufferIndexAndChannel> targetBuffers = new ArrayDeque<>();
                    // traverse buffers in order.
                    allBuffers.forEach(
                            (bufferContext -> {
                                if (isBufferSatisfyStatus(
                                        bufferContext, spillStatus, consumeStatusWithId)) {
                                    targetBuffers.add(bufferContext.getBufferIndexAndChannel());
                                }
                            }));
                    return targetBuffers;
                });
    }
@Test
    void testGetBuffersSatisfyStatus() throws Exception {
        TestingMemoryDataManagerOperation memoryDataManagerOperation =
                TestingMemoryDataManagerOperation.builder()
                        .setRequestBufferFromPoolSupplier(() -> createBufferBuilder(RECORD_SIZE))
                        .build();
        HsSubpartitionMemoryDataManager subpartitionMemoryDataManager =
                createSubpartitionMemoryDataManager(memoryDataManagerOperation);
        HsSubpartitionConsumerMemoryDataManager consumer =
                subpartitionMemoryDataManager.registerNewConsumer(DEFAULT);
        final int numBuffers = 4;
        for (int i = 0; i < numBuffers; i++) {
            subpartitionMemoryDataManager.append(createRecord(i), DataType.DATA_BUFFER);
        }

        // spill buffer 1 and 2
        List<BufferIndexAndChannel> toStartSpilling =
                HybridShuffleTestUtils.createBufferIndexAndChannelsList(0, 1, 2);
        CompletableFuture<Void> spilledDoneFuture = new CompletableFuture<>();
        subpartitionMemoryDataManager.spillSubpartitionBuffers(toStartSpilling, spilledDoneFuture);

        // consume buffer 0, 1
        consumer.consumeBuffer(0, Collections.emptyList());
        consumer.consumeBuffer(1, Collections.emptyList());

        checkBufferIndex(
                subpartitionMemoryDataManager.getBuffersSatisfyStatus(SpillStatus.ALL, ALL_ANY),
                Arrays.asList(0, 1, 2, 3));
        checkBufferIndex(
                subpartitionMemoryDataManager.getBuffersSatisfyStatus(
                        SpillStatus.ALL,
                        fromStatusAndConsumerId(ConsumeStatus.CONSUMED, HsConsumerId.DEFAULT)),
                Arrays.asList(0, 1));
        checkBufferIndex(
                subpartitionMemoryDataManager.getBuffersSatisfyStatus(
                        SpillStatus.ALL,
                        fromStatusAndConsumerId(ConsumeStatus.NOT_CONSUMED, HsConsumerId.DEFAULT)),
                Arrays.asList(2, 3));
        checkBufferIndex(
                subpartitionMemoryDataManager.getBuffersSatisfyStatus(SpillStatus.SPILL, ALL_ANY),
                Arrays.asList(1, 2));
        checkBufferIndex(
                subpartitionMemoryDataManager.getBuffersSatisfyStatus(
                        SpillStatus.NOT_SPILL, ALL_ANY),
                Arrays.asList(0, 3));
        checkBufferIndex(
                subpartitionMemoryDataManager.getBuffersSatisfyStatus(
                        SpillStatus.SPILL,
                        fromStatusAndConsumerId(ConsumeStatus.NOT_CONSUMED, HsConsumerId.DEFAULT)),
                Collections.singletonList(2));
        checkBufferIndex(
                subpartitionMemoryDataManager.getBuffersSatisfyStatus(
                        SpillStatus.SPILL,
                        fromStatusAndConsumerId(ConsumeStatus.CONSUMED, HsConsumerId.DEFAULT)),
                Collections.singletonList(1));
        checkBufferIndex(
                subpartitionMemoryDataManager.getBuffersSatisfyStatus(
                        SpillStatus.NOT_SPILL,
                        fromStatusAndConsumerId(ConsumeStatus.CONSUMED, HsConsumerId.DEFAULT)),
                Collections.singletonList(0));
        checkBufferIndex(
                subpartitionMemoryDataManager.getBuffersSatisfyStatus(
                        SpillStatus.NOT_SPILL,
                        fromStatusAndConsumerId(ConsumeStatus.NOT_CONSUMED, HsConsumerId.DEFAULT)),
                Collections.singletonList(3));
    }