private static void configureSerializationSchema(
            ReadableConfig formatOptions,
            KvRowDataSerializationSchema.Builder schemaBuilder) {

        formatOptions.getOptional(TextFormatOptions.CHARSET)
                .ifPresent(schemaBuilder::setCharset);

        formatOptions.getOptional(TextFormatOptions.KV_ENTRY_DELIMITER)
                .map(delimiter -> StringEscapeUtils.unescapeJava(delimiter).charAt(0))
                .ifPresent(schemaBuilder::setEntryDelimiter);

        formatOptions.getOptional(TextFormatOptions.KV_DELIMITER)
                .map(delimiter -> StringEscapeUtils.unescapeJava(delimiter).charAt(0))
                .ifPresent(schemaBuilder::setKvDelimiter);

        formatOptions
                .getOptional(ESCAPE_CHARACTER)
                .map(escape -> escape.charAt(0))
                .ifPresent(schemaBuilder::setEscapeCharacter);

        formatOptions
                .getOptional(QUOTE_CHARACTER)
                .map(quote -> quote.charAt(0))
                .ifPresent(schemaBuilder::setQuoteCharacter);

        formatOptions.getOptional(NULL_LITERAL)
                .ifPresent(schemaBuilder::setNullLiteral);

        formatOptions.getOptional(IGNORE_ERRORS)
                .ifPresent(schemaBuilder::setIgnoreErrors);

    }
Table format factory for providing configured instances of KV-to-RowData serializer and
deserializer.
inlong-sort/sort-formats/format-rowdata/format-rowdata-kv/src/main/java/org/apache/inlong/sort/formats/kv/KvFormatFactory.java