@SuppressWarnings("unchecked")
  protected final Schema doGenerate(TypeToken<?> typeToken, Set<String> knownRecords, boolean acceptRecursion)
    throws UnsupportedTypeException {
    Type type = typeToken.getType();
    Class<?> rawType = typeToken.getRawType();

    if (SIMPLE_SCHEMAS.containsKey(rawType)) {
      return SIMPLE_SCHEMAS.get(rawType);
    }

    // Enum type, simply use all the enum constants for ENUM schema.
    if (rawType.isEnum()) {
      return Schema.enumWith((Class<Enum<?>>) rawType);
    }

    // Java array, use ARRAY schema.
    if (rawType.isArray()) {
      Schema componentSchema = doGenerate(TypeToken.of(rawType.getComponentType()), knownRecords, acceptRecursion);
      if (rawType.getComponentType().isPrimitive()) {
        return Schema.arrayOf(componentSchema);
      }
      return Schema.arrayOf(Schema.unionOf(componentSchema, Schema.of(Schema.Type.NULL)));
    }

    if (!(type instanceof Class || type instanceof ParameterizedType)) {
      throw new UnsupportedTypeException("Type " + type + " is not supported. " +
                                         "Only Class or ParameterizedType are supported.");
    }

    // Any parameterized Collection class would be represented by ARRAY schema.
    if (Collection.class.isAssignableFrom(rawType)) {
      if (!(type instanceof ParameterizedType)) {
        throw new UnsupportedTypeException("Only supports parameterized Collection type.");
      }
      TypeToken<?> componentType = typeToken.resolveType(((ParameterizedType) type).getActualTypeArguments()[0]);
      Schema componentSchema = doGenerate(componentType, knownRecords, acceptRecursion);
      return Schema.arrayOf(Schema.unionOf(componentSchema, Schema.of(Schema.Type.NULL)));
    }

    // Java Map, use MAP schema.
    if (Map.class.isAssignableFrom(rawType)) {
      if (!(type instanceof ParameterizedType)) {
        throw new UnsupportedTypeException("Only supports parameterized Map type.");
      }
      Type[] typeArgs = ((ParameterizedType) type).getActualTypeArguments();
      TypeToken<?> keyType = typeToken.resolveType(typeArgs[0]);
      TypeToken<?> valueType = typeToken.resolveType(typeArgs[1]);

      Schema valueSchema = doGenerate(valueType, knownRecords, acceptRecursion);

      return Schema.mapOf(doGenerate(keyType, knownRecords, acceptRecursion),
                          Schema.unionOf(valueSchema, Schema.of(Schema.Type.NULL)));
    }

    // Any Java class, class name as the record name.
    String recordName = typeToken.getRawType().getName();
    if (knownRecords.contains(recordName)) {
      // Record already seen before
      if (acceptRecursion) {
        // simply create a reference RECORD schema by the name.
        return Schema.recordOf(recordName);
      } else {
        throw new UnsupportedTypeException("Recursive type not supported for class " + recordName);
      }
    }

    // Delegate to child class to generate RECORD schema.
    return generateRecord(typeToken, knownRecords, acceptRecursion);
  }
@ SuppressWarnings ( "unchecked" ) protected final Schema doGenerate ( TypeToken < ? > typeToken , Set < String > knownRecords , boolean acceptRecursion ) throws UnsupportedTypeException { Type type = typeToken . getType ( ) ; Class < ? > rawType = typeToken . getRawType ( ) ; if ( SIMPLE_SCHEMAS . containsKey ( rawType ) ) { return SIMPLE_SCHEMAS . get ( rawType ) ; } if ( rawType . isEnum ( ) ) { return Schema . enumWith ( ( Class < Enum < ? > > ) rawType ) ; } if ( rawType . isArray ( ) ) { Schema componentSchema = doGenerate ( TypeToken . of ( rawType . getComponentType ( ) ) , knownRecords , acceptRecursion ) ; if ( rawType . getComponentType ( ) . isPrimitive ( ) ) { return Schema . arrayOf ( componentSchema ) ; } return Schema . arrayOf ( Schema . unionOf ( componentSchema , Schema . of ( Schema . Type . NULL ) ) ) ; } if ( ! ( type instanceof Class || type instanceof ParameterizedType ) ) { throw new UnsupportedTypeException ( "Type " + type + " is not supported. " + "Only Class or ParameterizedType are supported." ) ; } if ( Collection . class . isAssignableFrom ( rawType ) ) { if ( ! ( type instanceof ParameterizedType ) ) { throw new UnsupportedTypeException ( "Only supports parameterized Collection type." ) ; } TypeToken < ? > componentType = typeToken . resolveType ( ( ( ParameterizedType ) type ) . getActualTypeArguments ( ) [ 0 ] ) ; Schema componentSchema = doGenerate ( componentType , knownRecords , acceptRecursion ) ; return Schema . arrayOf ( Schema . unionOf ( componentSchema , Schema . of ( Schema . Type . NULL ) ) ) ; } if ( Map . class . isAssignableFrom ( rawType ) ) { if ( ! ( type instanceof ParameterizedType ) ) { throw new UnsupportedTypeException ( "Only supports parameterized Map type." ) ; } Type [ ] typeArgs = ( ( ParameterizedType ) type ) . getActualTypeArguments ( ) ; TypeToken < ? > keyType = typeToken . resolveType ( typeArgs [ 0 ] ) ; TypeToken < ? > valueType = typeToken . resolveType ( typeArgs [ 1 ] ) ; Schema valueSchema = doGenerate ( valueType , knownRecords , acceptRecursion ) ; return Schema . mapOf ( doGenerate ( keyType , knownRecords , acceptRecursion ) , Schema . unionOf ( valueSchema , Schema . of ( Schema . Type . NULL ) ) ) ; } String recordName = typeToken . getRawType ( ) . getName ( ) ; if ( knownRecords . contains ( recordName ) ) { if ( acceptRecursion ) { return Schema . recordOf ( recordName ) ; } else { throw new UnsupportedTypeException ( "Recursive type not supported for class " + recordName ) ; } } return generateRecord ( typeToken , knownRecords , acceptRecursion ) ; }
Actual schema generation. It recursively resolves container types.

@param typeToken    Encapsulate the Java type for generating a {@link Schema}.
@param knownRecords Set of record names that has the schema already generated. It is used for
recursive class field references.
@param acceptRecursion Whether to tolerate type recursion. If false, will throw UnsupportedTypeException if
a recursive type is encountered.
@return A {@link Schema} representing the given java {@link java.lang.reflect.Type}.
@throws UnsupportedTypeException Indicates schema generation is not support for the
given java {@link java.lang.reflect.Type}.