class FileSourceSplitSerializer implements SimpleVersionedSerializer<FileSourceSplit> {

	@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());
		split.path().write(out);
		out.writeLong(split.offset());
		out.writeLong(split.length());
		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;
	}

}