public final class Schema implements Serializable {
private final Type type;
@Nullable
private final LogicalType logicalType;


public enum Type {
  NULL(true),
  BOOLEAN(true),
...
}


public enum LogicalType {
  TIME_MILLIS, //primitive type INT
  TIME_MICROS, // primitive type LONG
  TIMESTAMP_MILLIS, // primitive type LONG
  TIMESTAMP_MICROS, // primitive type LONG
  DATE // primitive type INT
}

/**
 * Creates a {@link Schema} for the given logical type. 
 *
 * @param type LogicalType of the schema to create.
 * @return A {@link Schema} with the given type.
 */
public static Schema of(LogicalType type) {
  if (type.equals(LogicalType.DATE)) {
    return new Schema(Type.INT, type, null, null, null, null, null, null, null);
  } else if (type.equals(LogicalType.TIMESTAMP_MILLIS)) {
    return new Schema(Type.LONG, type, null, null, null, null, null, null, null);
  }....
}

private Schema(Type type,@Nullable LogicalType logicalType,
               @Nullable Set<String> enumValues,                                    // Not null for enum type
               @Nullable Schema componentSchema,                                    // Not null for array type
               @Nullable Schema keySchema, @Nullable Schema valueSchema,            // Not null for map type
               @Nullable String recordName, @Nullable Map<String, Field> fieldMap,  // Not null for record type
               @Nullable List<Schema> unionSchemas) {                               // Not null for union type
  this.type = type;
  this.logicalType = logicalType;
  ...
}


/**
* @return The {@link LogicalType} that this schema represents.
*/
public LogicalType getLogicalType() {
 return logicalType;
}
}

public class SchemaExample {
  Schema schema = Schema.recordOf(
    "logicalTypeRecord",
    Schema.Field.of("id", Schema.of(Schema.Type.STRING)),
    Schema.Field.of("name", Schema.of(Schema.Type.STRING)),
    Schema.Field.of("date", Schema.of(Schema.LogicalType.DATE)),
    Schema.Field.of("timestamp", Schema.of(Schema.LogicalType.TIMESTAMP_MILLIS)));
}}

Date:
{
  "properties": {
    "schema": {
      "type": "record",
      "name": "etlSchemaBody",
      "fields": [
        {
          "name": "offset",
          "type": "long"
        },
        {
          "name": "body",
          "type": "string"
        },
        {
          "name": "Date",
          "type": {
            "type": "long",
            "logicalType": "date"
          }
        },
        {
          "name": "map",
          "type": {
            "type": "map",
            "keys": "long",
            "values": "float"
          }
        }
      ]
    }
  }
..........
}

Timestamp in micros:
{
  "properties": {
    "schema": {
      "type": "record",
      "name": "etlSchemaBody",
      "fields": [
        {
          "name": "offset",
          "type": "long"
        },
        {
          "name": "body",
          "type": "string"
        },
        {
          "name": "Timestamp-Micros",
          "type": {
            "type": "long",
            "logicalType": "timestamp-micros"
          }
        },
        {
          "name": "map",
          "type": {
            "type": "map",
            "keys": "long",
            "values": "float"
          }
        }
      ]
    }
  }
.......
}