@Override
    public SerializationSchema<RowData> createRuntimeEncoder(Context context, DataType physicalDataType) {
        final List<WriteableMetadata> writeableMetadata =
                metadataKeys.stream()
                        .map(
                                k -> Stream.of(WriteableMetadata.values())
                                        .filter(rm -> rm.key.equals(k))
                                        .findFirst()
                                        .orElseThrow(IllegalStateException::new))
                        .collect(Collectors.toList());
        return new CanalJsonEnhancedSerializationSchema(
                physicalDataType,
                writeableMetadata,
                timestampFormat,
                mapNullKeyMode,
                mapNullKeyLiteral,
                encodeDecimalAsPlainNumber);
    }
{@link EncodingFormat} for Canal using JSON encoding.

different from flink:1.13.5. This can apply metadata, sink metadata into canal format
inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedEncodingFormat.java