@Override
    public byte[] serialize(PendingSplitsCheckpoint<T> checkpoint) throws IOException {
        checkArgument(
                checkpoint.getClass() == PendingSplitsCheckpoint.class,
                "Cannot serialize subclasses of PendingSplitsCheckpoint");

        // optimization: the splits lazily cache their own serialized form
        if (checkpoint.serializedFormCache != null) {
            return checkpoint.serializedFormCache;
        }

        final SimpleVersionedSerializer<T> splitSerializer = this.splitSerializer; // stack cache
        final Collection<T> splits = checkpoint.getSplits();
        final Collection<Path> processedPaths = checkpoint.getAlreadyProcessedPaths();

        final ArrayList<byte[]> serializedSplits = new ArrayList<>(splits.size());
        final ArrayList<byte[]> serializedPaths = new ArrayList<>(processedPaths.size());

        int totalLen =
                16; // four ints: magic, version of split serializer, count splits, count paths

        for (T split : splits) {
            final byte[] serSplit = splitSerializer.serialize(split);
            serializedSplits.add(serSplit);
            totalLen += serSplit.length + 4; // 4 bytes for the length field
        }

        for (Path path : processedPaths) {
            final byte[] serPath = path.toString().getBytes(StandardCharsets.UTF_8);
            serializedPaths.add(serPath);
            totalLen += serPath.length + 4; // 4 bytes for the length field
        }

        final byte[] result = new byte[totalLen];
        final ByteBuffer byteBuffer = ByteBuffer.wrap(result).order(ByteOrder.LITTLE_ENDIAN);
        byteBuffer.putInt(VERSION_1_MAGIC_NUMBER);
        byteBuffer.putInt(splitSerializer.getVersion());
        byteBuffer.putInt(serializedSplits.size());
        byteBuffer.putInt(serializedPaths.size());

        for (byte[] splitBytes : serializedSplits) {
            byteBuffer.putInt(splitBytes.length);
            byteBuffer.put(splitBytes);
        }

        for (byte[] pathBytes : serializedPaths) {
            byteBuffer.putInt(pathBytes.length);
            byteBuffer.put(pathBytes);
        }

        assert byteBuffer.remaining() == 0;

        // optimization: cache the serialized from, so we avoid the byte work during repeated
        // serialization
        checkpoint.serializedFormCache = result;

        return result;
    }
@Test
    void repeatedSerializationCaches() throws Exception {
        final PendingSplitsCheckpoint<FileSourceSplit> checkpoint =
                PendingSplitsCheckpoint.fromCollectionSnapshot(
                        Collections.singletonList(testSplit2()));

        final byte[] ser1 =
                new PendingSplitsCheckpointSerializer<>(FileSourceSplitSerializer.INSTANCE)
                        .serialize(checkpoint);
        final byte[] ser2 =
                new PendingSplitsCheckpointSerializer<>(FileSourceSplitSerializer.INSTANCE)
                        .serialize(checkpoint);

        assertThat(ser1).isSameAs(ser2);
    }