@Override
        public byte[] serialize(PendingFileRecoverable pendingFileRecoverable) {
            if (!(pendingFileRecoverable instanceof HadoopPathBasedPartFileWriter.HadoopPathBasedPendingFileRecoverable)) {
                throw new UnsupportedOperationException("Only HadoopPathBasedPendingFileRecoverable is supported.");
            }

            HadoopPathBasedPendingFileRecoverable hadoopRecoverable =
                    (HadoopPathBasedPendingFileRecoverable) pendingFileRecoverable;
            Path path = hadoopRecoverable.getTargetFilePath();
            Path inProgressPath = hadoopRecoverable.getTempFilePath();

            byte[] pathBytes = path.toUri().toString().getBytes(CHARSET);
            byte[] inProgressBytes = inProgressPath.toUri().toString().getBytes(CHARSET);

            byte[] targetBytes = new byte[12 + pathBytes.length + inProgressBytes.length + Long.BYTES];
            ByteBuffer bb = ByteBuffer.wrap(targetBytes).order(ByteOrder.LITTLE_ENDIAN);
            bb.putInt(MAGIC_NUMBER);
            bb.putInt(pathBytes.length);
            bb.put(pathBytes);
            bb.putInt(inProgressBytes.length);
            bb.put(inProgressBytes);
            bb.putLong(hadoopRecoverable.getSize());

            return targetBytes;
        }
check if source table schema changes

@param identifier hive database name and table name
@param writerFactory hive writer factory
@param schema hive field with flink types
@return if schema has changed
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopPathBasedPartFileWriter.java