class PendingSplitsCheckpointSerializer implements SimpleVersionedSerializer<PendingSplitsCheckpoint> {

	@Override
	public byte[] serialize(PendingSplitsCheckpoint 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 FileSourceSplitSerializer serializer = FileSourceSplitSerializer.INSTANCE;
		final Collection<FileSourceSplit> 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, count splits, count paths

		for (FileSourceSplit split : splits) {
			final byte[] serSplit = serializer.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(serializer.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;
	}

}