protected static Object convertTo(Schema toSchema, Schema fromSchema, Object value) throws DataException {
        if (value == null) {
            if (toSchema.isOptional()) {
                return null;
            }
            throw new DataException("Unable to convert a null value to a schema that requires a value");
        }
        switch (toSchema.type()) {
            case BYTES:
                return convertMaybeLogicalBytes(toSchema, value);
            case STRING:
                return convertToString(fromSchema, value);
            case BOOLEAN:
                return convertToBoolean(fromSchema, value);
            case INT8:
                return convertToByte(fromSchema, value);
            case INT16:
                return convertToShort(fromSchema, value);
            case INT32:
                return convertMaybeLogicalInteger(toSchema, fromSchema, value);
            case INT64:
                return convertMaybeLogicalLong(toSchema, fromSchema, value);
            case FLOAT32:
                return convertToFloat(fromSchema, value);
            case FLOAT64:
                return convertToDouble(fromSchema, value);
            case ARRAY:
                return convertToArray(toSchema, value);
            case MAP:
                return convertToMapInternal(toSchema, value);
            case STRUCT:
                return convertToStructInternal(toSchema, value);
        }
        throw new DataException("Unable to convert " + value + " (" + value.getClass() + ") to " + toSchema);
    }
Convert the value to the desired type.

@param toSchema   the schema for the desired type; may not be null
@param fromSchema the schema for the supplied value; may be null if not known
@return the converted value; null if the passed-in schema was optional, and the input value was null.
@throws DataException if the value could not be converted to the desired type
connect/api/src/main/java/org/apache/kafka/connect/data/Values.java