@Override
    public byte[] serialize(MySqlSplit split) throws IOException {
        if (split.isSnapshotSplit()) {
            final MySqlSnapshotSplit snapshotSplit = split.asSnapshotSplit();
            // optimization: the splits lazily cache their own serialized form
            if (snapshotSplit.serializedFormCache != null) {
                return snapshotSplit.serializedFormCache;
            }

            final DataOutputSerializer out = SERIALIZER_CACHE.get();
            out.writeInt(SNAPSHOT_SPLIT_FLAG);
            out.writeUTF(quote(snapshotSplit.getTableId()));
            out.writeUTF(snapshotSplit.splitId());
            out.writeUTF(snapshotSplit.getSplitKeyType().asSerializableString());

            final Object[] splitStart = snapshotSplit.getSplitStart();
            final Object[] splitEnd = snapshotSplit.getSplitEnd();
            // rowToSerializedString deals null case
            out.writeUTF(rowToSerializedString(splitStart));
            out.writeUTF(rowToSerializedString(splitEnd));
            writeBinlogPosition(snapshotSplit.getHighWatermark(), out);
            writeTableSchemas(snapshotSplit.getTableSchemas(), out);
            final byte[] result = out.getCopyOfBuffer();
            out.clear();
            // optimization: cache the serialized from, so we avoid the byte work during repeated
            // serialization
            snapshotSplit.serializedFormCache = result;
            return result;
        } else {
            final MySqlBinlogSplit binlogSplit = split.asBinlogSplit();
            // optimization: the splits lazily cache their own serialized form
            if (binlogSplit.serializedFormCache != null) {
                return binlogSplit.serializedFormCache;
            }
            final DataOutputSerializer out = SERIALIZER_CACHE.get();
            out.writeInt(BINLOG_SPLIT_FLAG);
            out.writeUTF(binlogSplit.splitId());
            out.writeUTF("");
            writeBinlogPosition(binlogSplit.getStartingOffset(), out);
            writeBinlogPosition(binlogSplit.getEndingOffset(), out);
            writeFinishedSplitsInfo(binlogSplit.getFinishedSnapshotSplitInfos(), out);
            writeTableSchemas(binlogSplit.getTableSchemas(), out);
            out.writeInt(binlogSplit.getTotalFinishedSplitSize());
            out.writeBoolean(binlogSplit.isSuspended());
            final byte[] result = out.getCopyOfBuffer();
            out.clear();
            // optimization: cache the serialized from, so we avoid the byte work during repeated
            // serialization
            binlogSplit.serializedFormCache = result;
            return result;
        }
    }
@Test
    public void testRepeatedSerializationCache() throws Exception {
        final MySqlSplit split =
                new MySqlSnapshotSplit(
                        TableId.parse("test_db.test_table"),
                        "test_db.test_table-0",
                        new RowType(
                                Collections.singletonList(
                                        new RowType.RowField("id", new BigIntType()))),
                        null,
                        new Object[] {99L},
                        null,
                        new HashMap<>());
        final byte[] ser1 = MySqlSplitSerializer.INSTANCE.serialize(split);
        final byte[] ser2 = MySqlSplitSerializer.INSTANCE.serialize(split);
        assertSame(ser1, ser2);
    }