SubtaskCommittableManager<CommT> merge(SubtaskCommittableManager<CommT> other) {
        checkArgument(other.getSubtaskId() == this.getSubtaskId());
        this.numExpectedCommittables += other.numExpectedCommittables;
        this.requests.addAll(other.requests);
        this.numDrained += other.numDrained;
        this.numFailed += other.numFailed;
        return this;
    }
@Test
    void testMerge() {
        final SubtaskCommittableManager<Integer> subtaskCommittableManager =
                new SubtaskCommittableManager<>(
                        Collections.singletonList(new CommitRequestImpl<>(1, METRIC_GROUP)),
                        5,
                        1,
                        2,
                        1,
                        2L,
                        METRIC_GROUP);
        subtaskCommittableManager.merge(
                new SubtaskCommittableManager<>(
                        Arrays.asList(
                                new CommitRequestImpl<>(2, METRIC_GROUP),
                                new CommitRequestImpl<>(3, METRIC_GROUP)),
                        10,
                        2,
                        3,
                        1,
                        2L,
                        METRIC_GROUP));
        assertThat(subtaskCommittableManager.getNumCommittables()).isEqualTo(11);
        assertThat(subtaskCommittableManager.getNumDrained()).isEqualTo(3);
        assertThat(subtaskCommittableManager.isFinished()).isFalse();
        assertThat(subtaskCommittableManager.getNumFailed()).isEqualTo(5);
        assertThat(subtaskCommittableManager.getPendingRequests()).hasSize(3);
    }