private RowDataToFieldConverter createRowConverter(RowType type) {
        final String[] fieldNames = type.getFieldNames().toArray(new String[0]);
        final LogicalType[] fieldTypes =
                type.getFields().stream()
                        .map(RowType.RowField::getType)
                        .toArray(LogicalType[]::new);
        final RowDataToFieldConverter[] fieldConverters =
                Arrays.stream(fieldTypes)
                        .map(this::createConverter)
                        .toArray(RowDataToFieldConverter[]::new);
        final int fieldCount = type.getFieldCount();
        final RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[fieldTypes.length];
        for (int i = 0; i < fieldCount; i++) {
            fieldGetters[i] = RowData.createFieldGetter(fieldTypes[i], i);
        }

        return inputObj -> {
            RowData rowData = (RowData) inputObj;
            Map<String, Object> output = new HashMap<>();
            for (int i = 0; i < fieldCount; i++) {
                String fieldName = fieldNames[i];
                try {
                    Object field = fieldGetters[i].getFieldOrNull(rowData);
                    output.put(fieldName, fieldConverters[i].convert(field));
                } catch (Throwable t) {
                    throw new RuntimeException(String.format("Fail to serialize at field: %s.", fieldName), t);
                }
            }
            return output;
        };
    }
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/RowDataToFieldConverters.java