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;
}
}

/**
 * Get the value of date field in the record.
 * @param fieldName field to get.
 * @return Java 8 {@link LocalDate}, to get Java 7 {@link Date}, please use get method
 */
public LocalDate getDate(String fieldName) {

}

/**
 * Get the value of time field in the record.
 * @param fieldName field to get.
 * @return Java 8 {@link LocalTime}, to get Java 7 {@link Date}, please use get method
 */
public LocalTime getTime(String fieldName) {

}

/**
 * Get the value of date field in the record. UTC by default.
 * @param fieldName field to get.
 * @return Java 8 {@link Instant}, to get Java 7 {@link Date}, please use get method
 */
public ZonedDateTime getTimestamp(String fieldName) {
  
}


/**
 * Get the value of date field in the record.
 * @param fieldName field to get.
 * @return Java 8 {@link Instant}, to get Java 7 {@link Date}, please use get method
 */
public ZonedDateTime getTimestamp(String fieldName, ZoneId zoneId) {
  
}
/**
 * Set the date field to the given value
 * @param fieldName Name of the field to set
 * @param value Value for the field
 * @return This builder
 * @throws UnexpectedFormatException if the field is not in the schema, or the field is not nullable but a null
 * value is given
 */
public Builder setDate(String fieldName, @Nullable LocalDate value) {

}

/**
 * Set the time field to the given value
 * @param fieldName Name of the field to set
 * @param value Value for the field
 * @return This builder
 * @throws UnexpectedFormatException if the field is not in the schema, or the field is not nullable but a null
 * value is given
 */
public Builder setTime(String fieldName, @Nullable LocalTime value) {

}

/**
 * Set the timestamp field to the given value
 * @param fieldName Name of the field to set
 * @param value Value for the field
 * @return This builder
 * @throws UnexpectedFormatException if the field is not in the schema, or the field is not nullable but a null
 * value is given
 */
public Builder setTimestamp(String fieldName, @Nullable ZonedDateTime value) {

}

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)));
}}

StructuredRecord.Builder builder = StructuredRecord.builder(schema);
builder.setDate(dateField1, LocalDate.now());

Date:
{
  "properties": {
    "schema": {
      "type": "record",
      "name": "etlSchemaBody",
      "fields": [
        {
          "name": "Timestampoffset",
          "type": {"long"
        },
     "type": "long",  {
          "logicalTypename": "timestamp-microsbody",
           }"type": "string"
        },
        {
          "name": "mapDate",
          "type": {
            "type": "mapint",
            "keyslogicalType": "longdate",
          }
 "values": "float"      },
    }    {
    }       ]"name": "map",
    }   } ....... }



Time:
{
  "properties" "type": {
    "schema": {       "type": "recordmap",
 
    "name": "etlSchemaBody",       "fieldskeys": [
   "long",
    {           "namevalues": "offsetfloat",
          "type": "long"}
        },
      ]
 {   }
  }
..........
}

Timestamp "name":micros:
{
  "body",properties": {
    "schema": {
      "type": "string"
"record",
      "name": "etlSchemaBody",
       },"fields": [
        {
          "name": "Timeoffset",
          "type": {"long"
        },
   "type": "long",     {
          "logicalTypename": "time-microsbody",
          }
"type": "string"
        },
        {
          "name": "mapTimestamp-Micro",
          "type": {
            "type": "maplong",
            "keyslogicalType": "long",
 timestamp-micros"
          }
        },
        {
          "valuesname": "float"map",
          "type": {
     }       "type":  }"map",
         ]
    }
  }
.......
}

parse-timestamp <column_name> <time-unit>