@Override
    public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(
            Context context,
            ReadableConfig formatOptions) {
        validateFactoryOptions(this, formatOptions);
        validateEncodingFormatOptions(formatOptions);

        return new EncodingFormat<SerializationSchema<RowData>>() {

            @Override
            public SerializationSchema<RowData> createRuntimeEncoder(
                    DynamicTableSink.Context context,
                    DataType physicalDataType) {
                RowType rowType = (RowType) physicalDataType.getLogicalType();
                JsonRowDataSerializationSchema.Builder builder = JsonRowDataSerializationSchema.builder(rowType);
                return builder
                        .setMapNullKeyMode(formatOptions.get(TextFormatOptions.MAP_NULL_KEY_MODE))
                        .setMapNullKeyLiteral(formatOptions.get(TextFormatOptions.MAP_NULL_KEY_LITERAL))
                        .setTimestampFormat(formatOptions.get(TextFormatOptions.TIMESTAMP_FORMAT))
                        .setCharset(formatOptions.get(TextFormatOptions.CHARSET))
                        .setIgnoreErrors(formatOptions.get(TableFormatOptions.IGNORE_ERRORS))
                        .setObjectMapperConfig(formatOptions)
                        .build();
            }

            @Override
            public ChangelogMode getChangelogMode() {
                return ChangelogMode.insertOnly();
            }
        };
    }
Table format factory for providing configured instances of JSON to RowData {@link
SerializationSchema} and {@link DeserializationSchema}.
inlong-sort/sort-formats/format-rowdata/format-rowdata-json/src/main/java/org/apache/inlong/sort/formats/json/JsonFormatFactory.java