void upsertSummary(CommittableSummary<CommT> summary) {
        SubtaskCommittableManager<CommT> existing =
                subtasksCommittableManagers.putIfAbsent(
                        summary.getSubtaskId(),
                        new SubtaskCommittableManager<>(
                                summary.getNumberOfCommittables(),
                                subtaskId,
                                summary.getCheckpointId().isPresent()
                                        ? summary.getCheckpointId().getAsLong()
                                        : null,
                                metricGroup));
        if (existing != null) {
            throw new UnsupportedOperationException(
                    "Currently it is not supported to update the CommittableSummary for a checkpoint coming from the same subtask. Please check the status of FLINK-25920");
        }
    }
@Test
    void testUpdateCommittableSummary() {
        final CheckpointCommittableManagerImpl<Integer> checkpointCommittables =
                new CheckpointCommittableManagerImpl<>(1, 1, 1L, METRIC_GROUP);
        checkpointCommittables.upsertSummary(new CommittableSummary<>(1, 1, 1L, 1, 0, 0));
        assertThatThrownBy(
                        () ->
                                checkpointCommittables.upsertSummary(
                                        new CommittableSummary<>(1, 1, 1L, 2, 0, 0)))
                .isInstanceOf(UnsupportedOperationException.class)
                .hasMessageContaining("FLINK-25920");
    }