Thanks guys. Given Flink 1.12 is not ready (e.g. not available in Maven
repo), I need to stick to 1.11.
Dawid,
For the code throwing "java.lang.Long cannot be cast to java.time.Instant",
The avro schema has:
union {null, timestamp_ms } eventTime = null;
The avro pojo does have the logical type conversion:
private static SpecificData MODEL$ = new SpecificData();
static {
MODEL$.addLogicalTypeConversion(new
org.apache.avro.data.TimeConversions.TimestampMillisConversion());
}
I don't see SpecificRecord#getConversions() you mentioned in avro repo.
The pojo code throws:
public void put(int field$, java.lang.Object value$) {
switch (field$) {
case 3: eventTime = (java.time.Instant)value$; break; // throw here
}
I will send the full avdl and pojo offline to you for a close look.
Regards
Lian
On Mon, Sep 21, 2020 at 12:31 AM Aljoscha Krettek <aljos...@apache.org>
wrote:
> Hi All,
>
> Avro was finally bumped in
> https://issues.apache.org/jira/browse/FLINK-18192.
>
> The implementers didn't see
> https://issues.apache.org/jira/browse/FLINK-12532, but it is also
> updated now.
>
> Best,
> Aljoscha
>
> On 21.09.20 08:04, Arvid Heise wrote:
> > Hi Lian,
> >
> > we had a similar discussion on [1].
> >
> > TL;DR you are using Avro 1.9.x while Flink still bundles Avro 1.8 [2]
> until
> > Hive bumps it [3]. In the thread, I gave some options to avoid running
> into
> > the issue.
> > The easiest fix is to use Avro 1.8.2 all the way, but you may run into
> [4]
> > if your logical type is nullable (which is not necessary in most cases).
> >
> > Still, I think it's time for us to revise the decision to wait for Hive
> to
> > bump and rather upgrade independently. Avro was for a long time stuck on
> > 1.8 but the project gained traction again in the past two years. On the
> > other hand, Hive seems to be rather slow to respond to that and we
> > shouldn't have a slow moving component block us to support a fast moving
> > component if it's such apparent that users want it.
> > @Aljoscha Krettek <aljos...@apache.org> could you please pick that
> topic up
> > and ping the respective maintainers?
> >
> > [1]
> >
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Reading-from-AVRO-files-td35850.html
> > [2] https://issues.apache.org/jira/browse/FLINK-12532
> > [3] https://issues.apache.org/jira/browse/HIVE-21737
> > [4] https://issues.apache.org/jira/browse/AVRO-1891
> >
> > On Sun, Sep 20, 2020 at 9:56 PM Lian Jiang <jiangok2...@gmail.com>
> wrote:
> >
> >> Thanks Dawid for proposing ConfluentRegistryDeserializationSchema. I am
> >> trying ConfluentRegistryAvroDeserializationSchema (if this is what you
> >> mean) but got "java.lang.Long cannot be cast to java.time.Instant". This
> >> may be caused by https://issues.apache.org/jira/browse/FLINK-11030.
> >> <https://issues.apache.org/jira/browse/FLINK-11030> Is there any
> progress
> >> for this JIRA? Thanks. Regards!
> >>
> >>
> >> Stacktrace:
> >> java.lang.ClassCastException: java.lang.Long cannot be cast to
> >> java.time.Instant
> >> at com.mycompany.mypayload.MetadataRecord.put(MetadataRecord.java:136)
> >> at org.apache.avro.generic.GenericData.setField(GenericData.java:795)
> >> at
> >>
> org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:139)
> >> at
> >>
> org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:237)
> >> at
> >>
> org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123)
> >> at
> >>
> org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:170)
> >> at
> >>
> org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151)
> >> at
> >>
> org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:178)
> >> at
> >>
> org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:136)
> >> at
> >>
> org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:237)
> >> at
> >>
> org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123)
> >> at
> >>
> org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:170)
> >> at
> >>
> org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151)
> >> at
> >>
> org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:144)
> >> at
> >>
> org.apache.flink.formats.avro.RegistryAvroDeserializationSchema.deserialize(RegistryAvroDeserializationSchema.java:74)
> >> at
> >>
> com.mycompany.serde.SpecificRecordSerDe.deserialize(SpecificRecordSerDe.java:89)
> >> at
> >>
> com.mycompany.serde.SpecificRecordSerDe.deserialize(SpecificRecordSerDe.java:16)
> >> at
> >>
> org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema.deserialize(KafkaDeserializationSchema.java:80)
> >> at
> >>
> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:181)
> >> at
> >>
> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:141)
> >> at
> >>
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755)
> >> at
> >>
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
> >> at
> >>
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
> >> at
> >>
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201)
> >>
> >>
> >>
> >> Code:
> >>
> >> import org.apache.avro.specific.SpecificRecord;
> >> import org.apache.flink.api.common.typeinfo.TypeInformation;
> >> import
> org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroDeserializationSchema;
> >> import
> org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroSerializationSchema;
> >> import org.apache.flink.streaming.connectors.kafka.KafkaContextAware;
> >> import
> org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
> >> import
> org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
> >> import org.apache.kafka.clients.consumer.ConsumerRecord;
> >> import org.apache.kafka.clients.producer.ProducerRecord;
> >>
> >> import javax.annotation.Nullable;
> >> import java.io.Serializable;
> >>
> >> public class SpecificRecordSerDe<T extends SpecificRecord> implements
> >> KafkaSerializationSchema<T>, KafkaContextAware<T>,
> KafkaDeserializationSchema<T>, Serializable {
> >>
> >> private final Class<T> tClass;
> >> private String topic; // for serializer
> >> private String subject; // for serializer
> >> private final String schemaRegistryUrl;
> >> private ConfluentRegistryAvroSerializationSchema<T> serializer;
> >> private ConfluentRegistryAvroDeserializationSchema<T> deserializer;
> >>
> >> private static final Object lock = new Object();
> >>
> >> public static <T> SpecificRecordSerDe forDeserializer(final
> Class<T> tClass, String schemaRegistryUrl) {
> >> return new SpecificRecordSerDe(tClass, schemaRegistryUrl);
> >> }
> >>
> >> public static <T> SpecificRecordSerDe forSerializer(final Class<T>
> tClass,
> >> String
> schemaRegistryUrl,
> >> final String
> topic,
> >> final String
> subject) {
> >> return new SpecificRecordSerDe(tClass, schemaRegistryUrl,
> topic, subject);
> >> }
> >>
> >> private SpecificRecordSerDe(final Class<T> tClass, String
> schemaRegistryUrl) {
> >> this.tClass = tClass;
> >> this.schemaRegistryUrl = schemaRegistryUrl;
> >> }
> >>
> >> private SpecificRecordSerDe(final Class<T> tClass,
> >> final String schemaRegistryUrl,
> >> final String topic,
> >> final String subject) {
> >> this(tClass, schemaRegistryUrl);
> >> this.topic = topic;
> >> this.subject = subject;
> >> }
> >>
> >> @Override
> >> public ProducerRecord<byte[], byte[]> serialize(T element,
> @Nullable Long timestamp) {
> >> if (this.serializer == null) {
> >> synchronized (lock) {
> >> if (this.serializer == null) {
> >> this.serializer =
> ConfluentRegistryAvroSerializationSchema
> >> .forSpecific(tClass, this.subject,
> this.schemaRegistryUrl);
> >> }
> >> }
> >> }
> >>
> >> byte[] bytes = this.serializer.serialize(element);
> >> return new ProducerRecord<>(this.topic, bytes);
> >> }
> >>
> >> public boolean isEndOfStream(T nextElement) {
> >> return false;
> >> }
> >>
> >> @Override
> >> public T deserialize(ConsumerRecord<byte[], byte[]> record) throws
> Exception {
> >> if (deserializer == null) {
> >> synchronized (lock) {
> >> if (deserializer == null) {
> >> deserializer =
> ConfluentRegistryAvroDeserializationSchema
> >> .forSpecific(tClass,
> this.schemaRegistryUrl);
> >> }
> >> }
> >> }
> >>
> >> return deserializer.deserialize(record.value());
> >> }
> >>
> >> @Override
> >> public String getTargetTopic(T element) {
> >> return this.topic;
> >> }
> >>
> >> @Override
> >> public TypeInformation<T> getProducedType() {
> >> return TypeInformation.of(tClass);
> >> }
> >> }
> >>
> >>
> >>
> >>
> >> On Thu, Sep 17, 2020 at 9:42 AM Dawid Wysakowicz <
> dwysakow...@apache.org>
> >> wrote:
> >>
> >>> Thanks for the update.
> >>>
> >>> First of all, why did you decide to build your own
> DeserializationSchema
> >>> instead of using ConfluentRegistryDeserializationSchema? Your
> >>> implementation is quite inefficient you do deserialize > serialize >
> >>> deserialize. Serialization/deserialization is usually one of the
> heaviest
> >>> operations in the pipeline.
> >>>
> >>> What do you return in your getProducedType? From the stack trace I
> guess
> >>> you are instantiating the AvroTypeInfo? Could you maybe share a full
> >>> runnable example? It would make it much easier to help you.
> >>>
> >>> Moreover the pattern with registering custom conversions in a
> >>> SpecificData will not work with AvroSerializer. Custom serializers
> should
> >>> be defined in the generated SpecificRecord (in your case
> PayloadRecord) in
> >>> the SpecificRecordBase#getConversion().
> >>>
> >>> Best,
> >>>
> >>> Dawid
> >>>
> >>>
> >>> On 17/09/2020 16:34, Lian Jiang wrote:
> >>>
> >>> Piotr/Dawid,
> >>>
> >>> Thanks for the reply. FLINK-18223 seems not to related to this issue
> and
> >>> I double checked that I am using Flink 1.11.0 instead of 1.10.0. My
> >>> mistake. StreamExecutionEnvironment#getConfig()#enableObjectReuse())
> solved
> >>> the issue.
> >>>
> >>> I am not using ConfluentRegistryDeserializationSchema. Instead, I am
> >>> creating custom DeserializationSchema:
> >>>
> >>>
> >>> /*
> >>> the deser class
> >>> */
> >>> public class SpecificRecordSerDe<T extends SpecificRecord> implements
> >>>
> >>> KafkaSerializationSchema<T>, KafkaContextAware<T>,
> KafkaDeserializationSchema<T>, Serializable {
> >>> private final Class<T> tClass;private final String tSchemaStr;private
> volatile transient Schema tSchema;private String topic;private String
> schemaRegistryUrl;private KafkaAvroSerializer serializer;private
> KafkaAvroDecoder decoder;
> >>> public SpecificRecordSerDe(final Class<T> tClass, String tSchemaStr,
> String schemaRegistryUrl) {
> >>> this.tClass = tClass;
> >>> this.tSchemaStr = tSchemaStr;
> >>> this.topic = null;
> >>> this.schemaRegistryUrl = schemaRegistryUrl;
> >>> }
> >>>
> >>> @Overridepublic T deserialize(ConsumerRecord<byte[], byte[]> record)
> throws Exception { CachedSchemaRegistryClient client = new
> CachedSchemaRegistryClient(
> >>> schemaRegistryUrl,
> >>> 4);
> >>> decoder = new KafkaAvroDecoder(client);
> >>> GenericRecord generic = (GenericRecord)
> decoder.fromBytes(record.value()); DatumWriter<GenericRecord> writer =
> new SpecificDatumWriter<>(generic.getSchema(),
> ManagedSpecificData.getForClass(tClass));
> >>> ByteArrayOutputStream out = new ByteArrayOutputStream();
> >>> Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
> >>> writer.write(generic, encoder);
> >>> encoder.flush();
> >>>
> >>> byte[] avroData = out.toByteArray();
> >>> out.close();
> >>> tSchema = new org.apache.avro.Schema.Parser().parse(tSchemaStr);
> >>> SpecificDatumReader<T> reader = new SpecificDatumReader<>(
> >>> generic.getSchema(), tSchema,
> ManagedSpecificData.getForClass(tClass));
> >>> Decoder anotherDecoder =
> DecoderFactory.get().binaryDecoder(avroData, null);
> >>> T res = reader.read(null, anotherDecoder);
> >>>
> >>> return res;
> >>> }
> >>> }
> >>>
> >>>
> >>> /*
> >>> the specificData class
> >>> */public class ManagedSpecificData extends SpecificData { private
> static ManagedSpecificData getManagedSpecificData() {
> >>> ManagedSpecificData res = new ManagedSpecificData();
> >>>
> >>> registerAdvancedType(new TimestampMillisType(), res);
> >>> registerAdvancedType(new LocalDateType(), res);
> >>>
> >>> return res;
> >>> }}
> >>>
> >>>
> >>> /*
> >>>
> >>> how we use above deser class
> >>> */
> >>>
> >>> SpecificRecordSerDe<PayloadRecord> deserializer = new
> SpecificRecordSerDe<>(
> >>> PayloadRecord.class,
> >>> PayloadRecord.getClassSchema().toString(),
> >>> this.schemaRegistry);
> >>> FlinkKafkaConsumer consumer = new FlinkKafkaConsumer(
> >>> this.inputTopic,
> >>> deserializer,
> >>> this.sourceSettings);
> >>>
> >>>
> >>>
> >>> Thanks
> >>>
> >>> Lian
> >>>
> >>>
> >>>
> >>>
> >>> On Thu, Sep 17, 2020 at 2:19 AM Dawid Wysakowicz <
> dwysakow...@apache.org>
> >>> wrote:
> >>>
> >>>> Hi,
> >>>>
> >>>> Could you share exactly how do you configure avro & kafka? Do you use
> >>>> Table API or DataStream API? Do you use the
> >>>> ConfluentRegistryDeserializationSchema that comes with Flink or did
> you
> >>>> built custom DeserializationSchema? Could you maybe share the code for
> >>>> instantiating the source with us? It could help us track down the
> >>>> problematic spot.
> >>>>
> >>>> Best,
> >>>>
> >>>> Dawid
> >>>>
> >>>> On 16/09/2020 08:09, Lian Jiang wrote:
> >>>>> Hi,
> >>>>>
> >>>>> i am using avro 1.9.1 + Flink 1.10.1 + Confluent Kafka 5.5. In
> >>>>> Intellij, I can see the FlinkKafkaConsumer already deserialized the
> >>>>> upstream kafka message. However, I got below error when this message
> >>>>> is serialized during pushToOperator. Per the stack trace, the reason
> >>>>> is that AvroSerializer is created by AvroFactory.fromSpecific() which
> >>>>> creates its private copy of specificData. This private specificData
> >>>>> does not have logical type information. This blocks the deserialized
> >>>>> messages from being passed to downstream operators. Any idea how to
> >>>>> make this work? Appreciated very much!
> >>>>>
> >>>>>
> >>>>> org.apache.avro.AvroRuntimeException: Unknown datum type
> >>>>> java.time.Instant: 2020-09-15T07:00:00Z
> >>>>> at
> >>>>
> org.apache.avro.generic.GenericData.getSchemaName(GenericData.java:887)
> >>>>> at
> >>>>>
> >>>>
> org.apache.avro.specific.SpecificData.getSchemaName(SpecificData.java:420)
> >>>>> at
> >>>> org.apache.avro.generic.GenericData.resolveUnion(GenericData.java:850)
> >>>>> at
> >>>> org.apache.avro.generic.GenericData.deepCopyRaw(GenericData.java:1280)
> >>>>> at
> org.apache.avro.generic.GenericData.deepCopy(GenericData.java:1199)
> >>>>> at
> >>>> org.apache.avro.generic.GenericData.deepCopyRaw(GenericData.java:1261)
> >>>>> at
> org.apache.avro.generic.GenericData.deepCopy(GenericData.java:1199)
> >>>>> at
> >>>> org.apache.avro.generic.GenericData.deepCopyRaw(GenericData.java:1280)
> >>>>> at
> org.apache.avro.generic.GenericData.deepCopy(GenericData.java:1199)
> >>>>> at
> >>>> org.apache.avro.generic.GenericData.deepCopyRaw(GenericData.java:1261)
> >>>>> at
> org.apache.avro.generic.GenericData.deepCopy(GenericData.java:1199)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.formats.avro.typeutils.AvroSerializer.copy(AvroSerializer.java:242)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:715)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordsWithTimestamps(AbstractFetcher.java:352)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:185)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:141)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755)
> >>>>> at
> >>>>>
> >>>>
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
> >>>>
> >>>>
> >>>
> >>> --
> >>>
> >>>
> >>>
> >>> Create your own email signature
> >>> <
> https://www.wisestamp.com/signature-in-email/?utm_source=promotion&utm_medium=signature&utm_campaign=create_your_own&srcid=5234462839406592
> >
> >>>
> >>>
> >>
> >> --
> >>
> >> Create your own email signature
> >> <
> https://www.wisestamp.com/signature-in-email/?utm_source=promotion&utm_medium=signature&utm_campaign=create_your_own&srcid=5234462839406592
> >
> >>
> >
> >
>
>
--
Create your own email signature
<https://www.wisestamp.com/signature-in-email/?utm_source=promotion&utm_medium=signature&utm_campaign=create_your_own&srcid=5234462839406592>
