@Override
    public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
            DynamicTableFactory.Context context, ReadableConfig formatOptions) {
        FactoryUtil.validateFactoryOptions(this, formatOptions);
        JsonFormatOptionsUtil.validateDecodingFormatOptions(formatOptions);

        final String database = formatOptions.getOptional(DATABASE_INCLUDE).orElse(null);
        final String table = formatOptions.getOptional(TABLE_INCLUDE).orElse(null);
        final boolean ignoreParseErrors = formatOptions.get(IGNORE_PARSE_ERRORS);
        final TimestampFormat timestampFormat = JsonFormatOptionsUtil.getTimestampFormat(formatOptions);

        return new CanalJsonEnhancedDecodingFormat(database, table, ignoreParseErrors, timestampFormat);
    }
Format factory for providing configured instances of Canal JSON to RowData {@link
DeserializationSchema}.
Different from flink:1.13.5.This can sink metadata.
inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedFormatFactory.java