@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 testGetNumBytesProduced() {
        PointwiseBlockingResultInfo resultInfo =
                new PointwiseBlockingResultInfo(new IntermediateDataSetID(), 2, 2);
        resultInfo.recordPartitionInfo(0, new ResultPartitionBytes(new long[] {32L, 32L}));
        resultInfo.recordPartitionInfo(1, new ResultPartitionBytes(new long[] {64L, 64L}));

        assertThat(resultInfo.getNumBytesProduced()).isEqualTo(192L);
    }