private FieldToRowDataConverter createMapConverterByObjectMapper(
            String typeSummary,
            LogicalType keyType,
            LogicalType valueType) {
        if (!keyType.is(LogicalTypeFamily.CHARACTER_STRING)) {
            throw new UnsupportedOperationException(
                    "Currently we don't support non-string as key type of map. The type is: " + typeSummary);
        }
        final FieldToRowDataConverter keyConverter = createConverter(keyType);
        final FieldToRowDataConverter valueConverter = createConverter(valueType);

        return origStr -> {
            JsonNode jsonNode = objectMapper.readTree(origStr);
            Iterator<Entry<String, JsonNode>> fields = jsonNode.fields();
            Map<Object, Object> result = new HashMap<>();
            while (fields.hasNext()) {
                Map.Entry<String, JsonNode> entry = fields.next();
                Object key = keyConverter.convert(entry.getKey());
                Object value = valueConverter.convert(getStringFromJsonNode(entry.getValue()));
                result.put(key, value);
            }
            return new GenericMapData(result);
        };
    }
Creates a runtime converter which assuming input object is not null.
inlong-sort/sort-formats/format-rowdata/format-rowdata-json/src/main/java/org/apache/inlong/sort/formats/json/FieldToRowDataConverters.java