public static <T> AvroSchema<T> of(SchemaDefinition<T> schemaDefinition) {
        if (schemaDefinition.getSchemaReaderOpt().isPresent() && schemaDefinition.getSchemaWriterOpt().isPresent()) {
            return new AvroSchema<>(schemaDefinition.getSchemaReaderOpt().get(),
                    schemaDefinition.getSchemaWriterOpt().get(), parseSchemaInfo(schemaDefinition, SchemaType.AVRO));
        }
        ClassLoader pojoClassLoader = null;
        if (schemaDefinition.getClassLoader() != null) {
            pojoClassLoader = schemaDefinition.getClassLoader();
        } else if (schemaDefinition.getPojo() != null) {
            pojoClassLoader = schemaDefinition.getPojo().getClassLoader();
        }

        return new AvroSchema<>(parseSchemaInfo(schemaDefinition, SchemaType.AVRO), pojoClassLoader);
    }
@Test
    public void testJodaTimeLogicalType() {
        AvroSchema<JodaTimeLogicalType> avroSchema = AvroSchema.of(SchemaDefinition.<JodaTimeLogicalType>builder()
                .withPojo(JodaTimeLogicalType.class).build());
        JodaTimeLogicalType schemaLogicalType = new JodaTimeLogicalType();
        schemaLogicalType.setTimestampMicros(System.currentTimeMillis()*1000);
        schemaLogicalType.setTimestampMillis(new DateTime("2019-03-26T04:39:58.469Z", ISOChronology.getInstanceUTC()));
        schemaLogicalType.setDate(LocalDate.now());
        schemaLogicalType.setTimeMicros(System.currentTimeMillis()*1000);
        schemaLogicalType.setTimeMillis(LocalTime.now().truncatedTo(ChronoUnit.MILLIS));

        byte[] bytes1 = avroSchema.encode(schemaLogicalType);
        Assert.assertTrue(bytes1.length > 0);

        JodaTimeLogicalType object1 = avroSchema.decode(bytes1);

        assertEquals(object1, schemaLogicalType);
    }