是否有我可以发送和接收日期类型的 Apache Avro . 我没能在这上面找到任何东西。我发现只有在模式中使用int和logicalType的日期。但这会导致接收器端出现另一个int。我还得把它改成日期。

我正在尝试从 Apache Kafka 卡夫卡消费者中的生产者和接受者。

如果没有其他方法,那么我必须始终将日期转换为int,然后返回消费者。这篇文章展示了如何做到这一点:

Get the number of days, weeks, and months, since Epoch in Java

序列化程序代码:-

@Override
    public byte[] serialize(String topic, T data) {
        try {
            byte[] result = null;

            if (data != null) {
                logger.debug("data='{}'" +  data);

                ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
                BinaryEncoder binaryEncoder =
                        EncoderFactory.get().binaryEncoder(byteArrayOutputStream, null);

                DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(data.getSchema());
                datumWriter.write(data, binaryEncoder);

                binaryEncoder.flush();
                byteArrayOutputStream.close();

                result = byteArrayOutputStream.toByteArray();
                byteArrayOutputStream.close();
                logger.debug("serialized data='{}'" +  DatatypeConverter.printHexBinary(result));
            }
            return result;
        } catch (IOException ex) {
            throw new SerializationException(
                    "Can't serialize data='" + data + "' for topic='" + topic + "'", ex);
        }
    }

Desiralizer代码:-

    @Override
    public T deserialize(String topic, byte[] data) {
        try {
            T result = null;

            if (data != null) {
                logger.debug("data='{}'" + DatatypeConverter.printHexBinary(data));

                DatumReader<GenericRecord> datumReader =
                        new SpecificDatumReader<>(targetType.newInstance().getSchema());
                Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);

                result = (T) datumReader.read(null, decoder);
                logger.debug("deserialized data='{}'" + result);                
            }
            return result;
        } catch (Exception ex) {
            throw new SerializationException(
                    "Can't deserialize data '" + Arrays.toString(data) + "' from topic '" + topic + "'", ex);
        }
    }

架构文件:-

{"namespace": "com.test",
  "type": "record",
  "name": "Measures",
  "fields": [  
    {"name": "transactionDate", "type": ["int", "null"], "logicalType" : "date" }
   ]
}

这两个类只是在生产者和消费者配置中定义为序列化器和反序列化器类。