private SchemaAndValue parseMap() {
            Map<Object, Object> result = new LinkedHashMap<>();
            SchemaMerger keySchema = new SchemaMerger();
            SchemaMerger valueSchema = new SchemaMerger();
            while (parser.hasNext()) {
                if (parser.canConsume(MAP_END_DELIMITER)) {
                    Schema mapSchema;
                    if (keySchema.hasCommonSchema() && valueSchema.hasCommonSchema()) {
                        mapSchema = SchemaBuilder.map(keySchema.schema(), valueSchema.schema()).build();
                        result = alignMapKeysAndValuesWithSchema(mapSchema, result);
                    } else if (keySchema.hasCommonSchema()) {
                        mapSchema = SchemaBuilder.mapWithNullValues(keySchema.schema());
                        result = alignMapKeysWithSchema(mapSchema, result);
                    } else {
                        mapSchema = SchemaBuilder.mapOfNull().build();
                    }
                    return new SchemaAndValue(mapSchema, result);
                }

                if (parser.canConsume(COMMA_DELIMITER)) {
                    throw new DataException("Unable to parse a map entry with no key or value: " + parser.original());
                }
                SchemaAndValue key = parse(true);
                if (key == null || key.value() == null) {
                    throw new DataException("Map entry may not have a null key: " + parser.original());
                } else if (!parser.canConsume(ENTRY_DELIMITER)) {
                    throw new DataException("Map entry is missing '" + ENTRY_DELIMITER
                            + "' at " + parser.position()
                            + " in " + parser.original());
                }
                SchemaAndValue value = parse(true);
                Object entryValue = value != null ? value.value() : null;
                result.put(key.value(), entryValue);

                parser.canConsume(COMMA_DELIMITER);
                keySchema.merge(key);
                valueSchema.merge(value);
            }
            // Missing either a comma or an end delimiter
            if (COMMA_DELIMITER.equals(parser.previous())) {
                throw new DataException("Map is missing element after ',': " + parser.original());
            }
            throw new DataException("Map is missing terminating '}': " + parser.original());
        }
Convert the specified value with the desired floating point type.

@param value  the value to be converted; may not be null
@param schema the schema for the current value type; may not be null
@param error  any previous error that should be included in an exception message; may be null
@return the double value after conversion; never null
@throws DataException if the value could not be converted to a double
connect/api/src/main/java/org/apache/kafka/connect/data/Values.java