@Override
    protected ResultSubpartitionView createSubpartitionView(
            int subpartitionId, BufferAvailabilityListener availabilityListener)
            throws IOException {
        checkState(!isReleased(), "ResultPartition already released.");

        // If data file is not readable, throw PartitionNotFoundException to mark this result
        // partition failed. Otherwise, the partition data is not regenerated, so failover can not
        // recover the job.
        if (!Files.isReadable(dataFilePath)) {
            throw new PartitionNotFoundException(getPartitionId());
        }
        // if broadcastOptimize is enabled, map every subpartitionId to the special broadcast
        // subpartition.
        subpartitionId = isBroadcastOnly ? BROADCAST_SUBPARTITION : subpartitionId;

        HsSubpartitionConsumer subpartitionConsumer =
                new HsSubpartitionConsumer(availabilityListener);
        HsConsumerId lastConsumerId = lastConsumerIds[subpartitionId];
        checkMultipleConsumerIsAllowed(lastConsumerId, hybridShuffleConfiguration);
        // assign a unique id for each consumer, now it is guaranteed by the value that is one
        // higher than the last consumerId's id field.
        HsConsumerId consumerId = HsConsumerId.newId(lastConsumerId);
        lastConsumerIds[subpartitionId] = consumerId;
        HsDataView diskDataView =
                fileDataManager.registerNewConsumer(
                        subpartitionId, consumerId, subpartitionConsumer);

        HsDataView memoryDataView =
                checkNotNull(memoryDataManager)
                        .registerNewConsumer(subpartitionId, consumerId, subpartitionConsumer);

        subpartitionConsumer.setDiskDataView(diskDataView);
        subpartitionConsumer.setMemoryDataView(memoryDataView);
        return subpartitionConsumer;
    }
@Test
    void testFullSpillingStrategyRegisterMultipleConsumer() throws Exception {
        final int numSubpartitions = 2;
        BufferPool bufferPool = globalPool.createBufferPool(2, 2);
        try (HsResultPartition partition =
                createHsResultPartition(
                        2,
                        bufferPool,
                        HybridShuffleConfiguration.builder(
                                        numSubpartitions, readBufferPool.getNumBuffersPerRequest())
                                .setSpillingStrategyType(
                                        HybridShuffleConfiguration.SpillingStrategyType.FULL)
                                .build())) {
            partition.createSubpartitionView(
                    new ResultSubpartitionIndexSet(0), new NoOpBufferAvailablityListener());
            assertThatNoException()
                    .isThrownBy(
                            () ->
                                    partition.createSubpartitionView(
                                            new ResultSubpartitionIndexSet(0),
                                            new NoOpBufferAvailablityListener()));
        }
    }