Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
Locked
Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
 
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
In reply to this post by Lian Jiang
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) |
signature.asc (849 bytes)
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
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:
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
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. [hidden email] could you please pick that topic up and ping the respective maintainers? On Sun, Sep 20, 2020 at 9:56 PM Lian Jiang <[hidden email]> wrote:
-- Arvid Heise | Senior Java Developer Follow us @VervericaData -- Join Flink Forward - The Apache Flink Conference Stream Processing | Event Driven | Real Time -- Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany -- Ververica GmbHRegistered at Amtsgericht Charlottenburg: HRB 158244 B Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji (Toni) Cheng |
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Hey Arvid, Just a quick comment to Arvid's mail for now. It should be safe to update the Avro version even if we've been declaring dependency on Avro 1.8.2 by default. Moreover up until now we do not bundle any version of Avro in any of the uber jars we ship. It is true we used Avro version 1.8.2 by default because that's the version that hadoop ships with (the hadoop distributions really bundle avro dependency as part of their binaries). As for the other issue, because Hadoop is no longer the most frequent environment Flink is run and as you said they are not the fastest with upgrading dependencies we decided to upgrade the default Avro version that Flink declares. From Flink 1.12 by default we depend on Avro 1.10 It has already been merged into master[1]. Still users should be able to downgrade the avro version if they need. (If they have specific records generated with older versions or they use hadoop.) @Lian Will look further into the issue. My suspicion though is
there is a problem with the Conversions your generated class
declares. In order for Flink to handle logical types correctly the
generated Avro class must return valid Conversions via
SpecificRecord#getConversions(). Could you share the avro schema
and the generated class? Without the full picture it will be hard
to track down the problem. Again best would be an example that I
could run. Best, Dawid [1] https://issues.apache.org/jira/browse/FLINK-18192 On 21/09/2020 08:04, Arvid Heise wrote:
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Hi Lian, Thank you for sending the full code for the pojo. It clarified a lot! I learnt that Avro introduced yet another mechanism for
retrieving conversions for logical types in Avro 1.9.x. I was not
aware they create a static SpecificData field with registered
logical conversions if a logical type is part of a union. That's
why I did not understand the parts of the you sent me where you
are registering the logical types in the MODEL$ field. The
getConversion method is part of the SpecificRecordBase class and
is being populated by Avro compiler when a logical type is a top
level type. This bit should work just fine. Unfortunately we do not support this "feature" of using the
static SpecificData field. So far we create the SpecificData
manually in AvroSerializer and Avro(De)SerializationSchema that is
why the conversions are not being picked up. I created a JIRA
issue[1] and a PR[2] to support it in Flink 1.12. The only workaround I can see in earlier versions of Flink is to change the AvroSerializer manually. You would need to do a similar thing as I do in the linked PR. Best, Dawid [1] https://issues.apache.org/jira/browse/FLINK-19339 [2] https://github.com/apache/flink/pull/13450 On 21/09/2020 19:28, Lian Jiang wrote:
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Hi Lian, sorry for the late reply. 1. All serialization related functions are just implementation of API interfaces. As such, you can implement serializers yourself. In this case, you could simply copy the code from 1.12 into your application. You may adjust a few things that are different between 1.11 and 1.12 though. 2. enableObjectReuse avoids copying of records between chained operators. The chain ends with any keyby. A possible workaround is to enableObjectReuse and then convert the datum with logical types into a datum without logical types if possible (potentially dropping a few fields to reduce network traffic along the way). On Wed, Sep 23, 2020 at 6:11 PM Lian Jiang <[hidden email]> wrote:
-- Arvid Heise | Senior Java Developer Follow us @VervericaData -- Join Flink Forward - The Apache Flink Conference Stream Processing | Event Driven | Real Time -- Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany -- Ververica GmbHRegistered at Amtsgericht Charlottenburg: HRB 158244 B Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji (Toni) Cheng |
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
Re: Unknown datum type java.time.Instant: 2020-09-15T07:00:00Z
|
|
| Free forum by Nabble | Edit this page |