[
https://issues.apache.org/jira/browse/FLINK-12577?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Flink Jira Bot updated FLINK-12577:
-----------------------------------
Labels: auto-deprioritized-major stale-minor (was:
auto-deprioritized-major)
I am the [Flink Jira Bot|https://github.com/apache/flink-jira-bot/] and I help
the community manage its development. I see this issues has been marked as
Minor but is unassigned and neither itself nor its Sub-Tasks have been updated
for 180 days. I have gone ahead and marked it "stale-minor". If this ticket is
still Minor, please either assign yourself or give an update. Afterwards,
please remove the label or in 7 days the issue will be deprioritized.
> Flink uses Kryo serializer if given a SpecificRecord type
> ---------------------------------------------------------
>
> Key: FLINK-12577
> URL: https://issues.apache.org/jira/browse/FLINK-12577
> Project: Flink
> Issue Type: Bug
> Components: API / Type Serialization System
> Affects Versions: 1.7.2, 1.8.0, 1.9.0, 1.10.0
> Reporter: Nico Kruber
> Priority: Minor
> Labels: auto-deprioritized-major, stale-minor
>
> If a data stream does not return a *subclass* of {{SpecificRecord}}, Flink
> will actually infer a {{GenericTypeInfo}} and use kryo instead.
> Example:
> {code}
> StreamExecutionEnvironment env = ...
> env.getConfig().disableGenericTypes();
> DataStream<Measurement> sourceStream = env.addSource(...);
> DataStream<SpecificRecord> test = sourceStream.map(value -> new
> AggregatedSensorStatistics());
> public class AggregatedSensorStatistics extends
> org.apache.avro.specific.SpecificRecordBase implements
> org.apache.avro.specific.SpecificRecord {...}
> {code}
> This will lead to
> {code}
> Exception in thread "main" java.lang.UnsupportedOperationException: Generic
> types have been disabled in the ExecutionConfig and type
> org.apache.avro.specific.SpecificRecord is treated as a generic type.
> at
> org.apache.flink.api.java.typeutils.GenericTypeInfo.createSerializer(GenericTypeInfo.java:86)
> at
> org.apache.flink.streaming.api.graph.StreamGraph.addOperator(StreamGraph.java:208)
> at
> org.apache.flink.streaming.api.graph.StreamGraphGenerator.transformOneInputTransform(StreamGraphGenerator.java:541)
> at
> org.apache.flink.streaming.api.graph.StreamGraphGenerator.transform(StreamGraphGenerator.java:166)
> at
> org.apache.flink.streaming.api.graph.StreamGraphGenerator.generateInternal(StreamGraphGenerator.java:132)
> at
> org.apache.flink.streaming.api.graph.StreamGraphGenerator.generate(StreamGraphGenerator.java:124)
> at
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:1537)
> at
> org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:89)
> at
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1510)
> at
> com.ververica.training.statemigration.StateMigrationJobBase.createAndExecuteJob(StateMigrationJobBase.java:68)
> at
> com.ververica.training.statemigration.avro.StateMigrationJob.main(StateMigrationJob.java:13)
> {code}
> You may want some flexibility in your types and thus not provide the exact
> one like {{AggregatedSensorStatistics}} in this example. I don't see any
> reason we should disallow that behaviour.
> Reason for this is that {{TypeExtractor#privateGetForClass()}} is having this
> code which only sees classes as Avro if they *extend* from {{SpecificRecord}}:
> {code}
> if (hasSuperclass(clazz, AVRO_SPECIFIC_RECORD_BASE_CLASS)) {
> return AvroUtils.getAvroUtils().createAvroTypeInfo(clazz);
> }
> {code}
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
