@Override
    public CommittableCollector<CommT> deserialize(int version, byte[] serialized)
            throws IOException {
        final DataInputDeserializer in = new DataInputDeserializer(serialized);
        if (version == 1) {
            return deserializeV1(in);
        }
        if (version == 2) {
            validateMagicNumber(in);
            return deserializeV2(in);
        }
        throw new IOException("Unrecognized version or corrupt state: " + version);
    }
@Test
    void testCommittableCollectorV1SerDe() throws IOException {
        final List<Integer> legacyState = Arrays.asList(1, 2, 3);
        final DataOutputSerializer out = new DataOutputSerializer(256);
        out.writeInt(SinkV1CommittableDeserializer.MAGIC_NUMBER);
        SimpleVersionedSerialization.writeVersionAndSerializeList(
                COMMITTABLE_SERIALIZER, legacyState, out);
        final byte[] serialized = out.getCopyOfBuffer();
        final CommittableCollector<Integer> committableCollector =
                SERIALIZER.deserialize(1, serialized);
        assertThat(committableCollector.getNumberOfSubtasks()).isEqualTo(1);
        assertThat(committableCollector.isFinished()).isFalse();
        assertThat(committableCollector.getSubtaskId()).isEqualTo(0);
        final Collection<CheckpointCommittableManagerImpl<Integer>> checkpointCommittables =
                committableCollector.getCheckpointCommittables();
        assertThat(checkpointCommittables).hasSize(1);
        final SubtaskCommittableManager<Integer> subtaskCommittableManager =
                checkpointCommittables.iterator().next().getSubtaskCommittableManager(0);
        assertThat(
                        subtaskCommittableManager
                                .getPendingRequests()
                                .map(CommitRequestImpl::getCommittable)
                                .collect(Collectors.toList()))
                .containsExactly(1, 2, 3);
    }