public Optional<ShuffleMetrics> getMetricsOfPartition(ResultPartitionID partitionId) {
        synchronized (registeredPartitions) {
            final ResultPartition partition = registeredPartitions.get(partitionId);

            if (partition == null) {
                return Optional.empty();
            }

            return Optional.of(
                    new DefaultShuffleMetrics(
                            partition.getResultPartitionBytes().createSnapshot()));
        }
    }
@Test
    void testGetMetricsOfPartition() throws Exception {
        final ResultPartitionManager partitionManager = new ResultPartitionManager();
        final ResultPartition partition = createPartition();
        partition.resultPartitionBytes.incAll(100);

        partitionManager.registerResultPartition(partition);
        Optional<ShuffleMetrics> metricsOfPartition =
                partitionManager.getMetricsOfPartition(partition.partitionId);
        assertThat(metricsOfPartition)
                .hasValueSatisfying(
                        metrics ->
                                Arrays.equals(
                                        metrics.getPartitionBytes().getSubpartitionBytes(),
                                        new long[] {100}));
    }