@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 testPartitionFinishedMultiTimes() {
        PointwiseBlockingResultInfo resultInfo =
                new PointwiseBlockingResultInfo(new IntermediateDataSetID(), 2, 2);

        resultInfo.recordPartitionInfo(0, new ResultPartitionBytes(new long[] {32L, 64L}));
        resultInfo.recordPartitionInfo(1, new ResultPartitionBytes(new long[] {64L, 128L}));
        assertThat(resultInfo.getNumOfRecordedPartitions()).isEqualTo(2);
        assertThat(resultInfo.getNumBytesProduced()).isEqualTo(288L);

        // reset partition info
        resultInfo.resetPartitionInfo(0);
        assertThat(resultInfo.getNumOfRecordedPartitions()).isOne();

        // record partition info again
        resultInfo.recordPartitionInfo(0, new ResultPartitionBytes(new long[] {64L, 128L}));
        assertThat(resultInfo.getNumBytesProduced()).isEqualTo(384L);
    }