@Override
    public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(Context context,
            ReadableConfig formatOptions) {
        validateDecodingFormatOptions(formatOptions);

        final DeserializationFormatFactory innerFactory = FactoryUtil.discoverFactory(
                context.getClassLoader(),
                DeserializationFormatFactory.class,
                formatOptions.get(INNER_FORMAT));
        Configuration allOptions = Configuration.fromMap(context.getCatalogTable().getOptions());
        String innerFormatMetaPrefix = formatOptions.get(INNER_FORMAT) + ".";
        String innerFormatPrefix = INLONG_PREFIX + innerFormatMetaPrefix;
        DecodingFormat<DeserializationSchema<RowData>> innerFormat =
                innerFactory.createDecodingFormat(context, new DelegatingConfiguration(allOptions, innerFormatPrefix));
        boolean ignoreErrors = formatOptions.get(IGNORE_PARSE_ERRORS);
        boolean ignoreTrailingUnmappable = formatOptions.get(IGNORE_TRAILING_UNMAPPABLE);
        String decompressType = formatOptions.get(DECOMPRESS_TYPE);
        return new InLongMsgPbDecodingFormat(innerFormat, innerFormatMetaPrefix,
                ignoreErrors, ignoreTrailingUnmappable, decompressType);
    }
factory class for inLong msg pb format
inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-pb/src/main/java/org/apache/inlong/sort/formats/inlongmsgpb/InLongMsgPbFormatFactory.java