@Override
    public byte[] serialize(FileSourceSplit split) throws IOException {
        checkArgument(
                split.getClass() == FileSourceSplit.class,
                "Cannot serialize subclasses of FileSourceSplit");

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

        final DataOutputSerializer out = SERIALIZER_CACHE.get();

        out.writeUTF(split.splitId());
        Path.serializeToDataOutputView(split.path(), out);
        out.writeLong(split.offset());
        out.writeLong(split.length());
        out.writeLong(split.fileModificationTime());
        out.writeLong(split.fileSize());
        writeStringArray(out, split.hostnames());

        final Optional<CheckpointedPosition> readerPosition = split.getReaderPosition();
        out.writeBoolean(readerPosition.isPresent());
        if (readerPosition.isPresent()) {
            out.writeLong(readerPosition.get().getOffset());
            out.writeLong(readerPosition.get().getRecordsAfterOffset());
        }

        final byte[] result = out.getCopyOfBuffer();
        out.clear();

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

        return result;
    }
@Test
    void repeatedSerializationCaches() throws Exception {
        final FileSourceSplit split =
                new FileSourceSplit(
                        "random-id",
                        new Path("hdfs://namenode:14565/some/path/to/a/file"),
                        100_000_000,
                        64_000_000,
                        System.currentTimeMillis(),
                        200_000_000,
                        "host1",
                        "host2",
                        "host3");

        final byte[] ser1 = FileSourceSplitSerializer.INSTANCE.serialize(split);
        final byte[] ser2 = FileSourceSplitSerializer.INSTANCE.serialize(split);

        assertThat(ser1).isSameAs(ser2);
    }