private void makeCsvInnerFormatIgnoreTrailingUnmappable(DeserializationSchema<RowData> innerSchema) {
        try {
            Field readerField = CsvRowDataDeserializationSchema.class.getDeclaredField("objectReader");
            readerField.setAccessible(true);
            ObjectReader oldReader = (ObjectReader) readerField.get(innerSchema);

            Field schemaField = ObjectReader.class.getDeclaredField("_schema");
            schemaField.setAccessible(true);
            CsvSchema oldSchema = (CsvSchema) schemaField.get(oldReader);

            ObjectReader newReader = new CsvMapper()
                    .enable(CsvParser.Feature.IGNORE_TRAILING_UNMAPPABLE)
                    .readerFor(JsonNode.class)
                    .with(oldSchema);
            readerField.set(innerSchema, newReader);
        } catch (Throwable t) {
            log.error("failed to make csv inner format to ignore trailing unmappable, ex is ", t);
        }
    }
Use reflection to make csv format ignore tailing unmappable.
inlong-sort/sort-formats/format-row/format-inlongmsg-pb/src/main/java/org/apache/inlong/sort/formats/inlongmsgpb/InLongMsgPbDecodingFormat.java