@Override
    public long getNumBytesProduced() {
        checkState(
                subpartitionBytesByPartitionIndex.size() == numOfPartitions,
                "Not all partition infos are ready");
        return subpartitionBytesByPartitionIndex.values().stream()
                .flatMapToLong(Arrays::stream)
                .reduce(0L, Long::sum);
    }
@Test
    void testGetNumBytesProducedWithIndexRange() {
        PointwiseBlockingResultInfo resultInfo =
                new PointwiseBlockingResultInfo(new IntermediateDataSetID(), 2, 2);
        resultInfo.recordPartitionInfo(0, new ResultPartitionBytes(new long[] {32L, 64L}));
        resultInfo.recordPartitionInfo(1, new ResultPartitionBytes(new long[] {128L, 256L}));

        IndexRange partitionIndexRange = new IndexRange(0, 0);
        IndexRange subpartitionIndexRange = new IndexRange(0, 1);

        assertThat(resultInfo.getNumBytesProduced(partitionIndexRange, subpartitionIndexRange))
                .isEqualTo(96L);
    }