{
"type": "record",
"name": "SomeMessage",
"namespace": "com.trthhrts",
"fields": [
   {
     "name": "title",
     "type": "string"
   }
  ]
}

{
"type": "record",
"name": "AnotherMessage",
"namespace": "com.trthhrts",
"fields": [
    {
     "name": "messageDate",
     "type": [
       "null",
       {
         "type": "long",
         "logicalType": "timestamp-millis"
       }
     ],
     "default": null
    }
  ]
}

<plugin>
   <groupId>io.confluent</groupId>
   <artifactId>kafka-schema-registry-maven-plugin</artifactId>
   <version>6.0.1</version>
   <configuration>
     <schemaRegistryUrls>
         <param>${SCHEMA_REGISTRY_URL}</param>
     </schemaRegistryUrls>
     <outputDirectory>${project.build}/avro</outputDirectory>
     <userInfoConfig />
     <subjects>
       <com.trthhrts.SomeMessage>src/SomeMessage.avsc</com.trthhrts.SomeMessage>
     </subjects>
     <schemaTypes>
       <com.trthhrts.SomeMessage>AVRO</com.trthhrts.SomeMessage>
     </schemaTypes>
   </configuration>
</plugin>

register_schemas:
stage: deploy
script:
   - mvn schema-registry:test-compatibility $MAVEN_CLI_OPTS -B
   - mvn schema-registry:register $MAVEN_CLI_OPTS -B
when: manual
only:
   - master

register_schemas:
stage: deploy
before_script:
   - *scm_connect
script:
   - mvn schema-registry:validate $MAVEN_CLI_OPTS -B
   - mvn schema-registry:test-compatibility $MAVEN_CLI_OPTS -B
   - mvn schema-registry:register $MAVEN_CLI_OPTS -B
when: manual
only:
   - master

props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);

@Bean
    public ConcurrentKafkaListenerContainerFactory<String, ? extends SpecificRecordBase> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, ? extends SpecificRecordBase> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(kafkaConfigs(), new StringDeserializer(), deserializer()));
        factory.setErrorHandler(errorHandler());
        return factory;
    }
    private KafkaAvroDeserializer deserializer() {
        KafkaAvroDeserializer deserializer = new KafkaAvroDeserializer();
        deserializer.configure(kafkaConfigs(), false);
        return deserializer;
    }

props.put(AbstractKafkaAvroSerDeConfig.VALUE_SUBJECT_NAME_STRATEGY, RecordNameStrategy.class);

props.put(AbstractKafkaAvroSerDeConfig.AUTO_REGISTER_SCHEMAS, false);

put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);

static {
  SpecificData.get().addLogicalTypeConversion(new Conversions.DecimalConversion());
  SpecificData.get().addLogicalTypeConversion(new TimeConversions.TimestampConversion());
}

{
  "type": "record",
  "name": "SomePaymentInfoModel",
  "namespace": "com.trthhrts",
  "fields": [
    {
      "name": "paymentDate",
      "type": [
        "null",
        {
          "type": "long",
          "logicalType": "timestamp-millis"
        }
      ],
      "default": null
    },
    {
      "name": "amount",
      "type": [
        "null",
        {
          "type": "bytes",
          "logicalType": "decimal",
          "precision": 9,
          "scale": 2
        }
      ],
      "default": null
    }
  ]
}

convertDate(model.getBeginDate()).ifPresent(builder::setBeginDate)
convertLocalDateToJodaDateTime(model.getPaymentDate()).ifPresent(builder::setPaymentDate);

convertToLocalDateTime(message.getLastUpdate()).ifPresent(model::setLastUpdate);

/**
* Класс с утилитными методам для конвертации типов
*/
public final class MessageUtils {
    /** Кол-во наносекунд в миллисекунде */
    private static final int NANO_SECONDS_IN_MILLISECOND = 1_000_000;
    /** Кол-во миллисекунд в секунде */
    private static final int MILLISECONDS_IN_SECONDS = 1000;
    /**
     * Запрещаем создание
     */
    private MessageUtils() {
    }
    /**
     * Конвертирует Joda {@link DateTime} в Java 8 {@link ZonedDateTime}
     * @param dateTime дата в формате Joda {@link DateTime}
     * @return {@link ZonedDateTime}
     */
    public static Optional<ZonedDateTime> convertToZonedDateTime(DateTime dateTime) {
        return Optional.ofNullable(dateTime).map(dt -> {
            DateTime localDateTime = dateTime.withZone(DateTimeZone.getDefault());
            return ZonedDateTime.ofLocal(
                    LocalDateTime.of(
                            localDateTime.getYear(),
                            localDateTime.getMonthOfYear(),
                            localDateTime.getDayOfMonth(),
                            localDateTime.getHourOfDay(),
                            localDateTime.getMinuteOfHour(),
                            localDateTime.getSecondOfMinute(),
                            localDateTime.getMillisOfSecond() * NANO_SECONDS_IN_MILLISECOND),
                    ZoneId.of(localDateTime.getZone().getID(), ZoneId.SHORT_IDS),
                    ZoneOffset.ofTotalSeconds(localDateTime.getZone().getOffset(localDateTime) / MILLISECONDS_IN_SECONDS));
        });
    }
    /**
     * Конвертирует Joda {@link DateTime} в Java 8 {@link LocalDateTime}
     * @param dateTime дата в формате Joda {@link DateTime}
     * @return {@link LocalDateTime}
     */
    public static Optional<LocalDateTime> convertToLocalDateTime(DateTime dateTime) {
        return convertToZonedDateTime(dateTime).map(ZonedDateTime::toLocalDateTime);
    }
    /**
     * Конвертирует Joda {@link DateTime} в Java 8 {@link LocalDate}
     * @param dateTime дата в формате Joda {@link DateTime}
     * @return {@link LocalDate}
     */
    public static Optional<LocalDate> convertToLocalDate(DateTime dateTime) {
        return convertToZonedDateTime(dateTime).map(ZonedDateTime::toLocalDate);
    }
    /**
     * Конвертирует {@link LocalDate} в {@link org.joda.time.LocalDate}
     * @param ld {@link LocalDate}
     * @return {@link org.joda.time.LocalDate}
     */
    public static Optional<org.joda.time.LocalDate> convertLocalDateToJodaLocalDate(LocalDate ld) {
        return Optional.ofNullable(ld).map(date -> new org.joda.time.LocalDate(date.getYear(), date.getMonthValue(), date.getDayOfMonth()));
    }
    /**
     * Конвертирует {@link org.joda.time.LocalDate} в {@link LocalDate}
     * @param ld {@link org.joda.time.LocalDate}
     * @return {@link LocalDate}
     */
    public static Optional<LocalDate> convertJodaLocalDateToLocalDate(org.joda.time.LocalDate ld) {
        return Optional.ofNullable(ld).map(date -> LocalDate.of(date.getYear(), date.getMonthOfYear(), date.getDayOfMonth()));
    }
    /**
     * Конвертирует {@link LocalDateTime} в Joda {@link DateTime}
     * @param ldt дата в формате {@link LocalDateTime}
     * @return дата в формате Joda {@link DateTime}
     */
    public static Optional<DateTime> convertLocalDateTimeToJodaDateTime(LocalDateTime ldt) {
        return Optional.ofNullable(ldt)
                .map(date -> new org.joda.time.LocalDateTime(date.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()).toDateTime());
    }
    /**
     * Конвертирует {@link LocalDate} в Joda {@link DateTime}
     * @param ld дата в формате {@link LocalDate}
     * @return дата в формате Joda {@link DateTime}
     */
    public static Optional<DateTime> convertLocalDateToJodaDateTime(LocalDate ld) {
        return Optional.ofNullable(ld)
                .map(date ->
                        new org.joda.time.LocalDate(date.getYear(), date.getMonthValue(), date.getDayOfMonth()).toDateTimeAtStartOfDay());
    }
}

{
  "type": "record",
  "name": "FromMaven",
  "namespace": "com.trthhrts",
  "fields": [
    {
      "name": "message",
      "type": {
        "type": "string",
        "avro.java.string": "String"
      }
    }
  ]
}