Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/323#discussion_r23435527
--- Diff:
flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
---
@@ -237,6 +244,25 @@ private void checkKryoInitialized() {
// Throwable and all subclasses should be serialized
via java serialization
kryo.addDefaultSerializer(Throwable.class, new
JavaSerializer());
+ // If the type we have to serialize as a GenricType is
implementing SpecificRecordBase,
+ // we have to register the avro serializer
+ // This rule only applies if users explicitly use the
GenericTypeInformation for the avro types
+ // usually, we are able to handle Avro POJOs with the
POJO serializer.
+ if(SpecificRecordBase.class.isAssignableFrom(type)) {
+ ClassTag<SpecificRecordBase> tag =
scala.reflect.ClassTag$.MODULE$.apply(type);
+ this.kryo.register(type,
com.twitter.chill.avro.AvroSerializer.SpecificRecordSerializer(tag));
+
+ }
+ // Avro POJOs contain java.util.List which have
GenericData.Array as their runtime type
+ // because Kryo is not able to serialize them properly,
we use this serializer for them
+ this.kryo.register(GenericData.Array.class, new
SpecificInstanceCollectionSerializer(ArrayList.class));
--- End diff --
Should we make this registration conditional so that it only happens when
we have encountered an Avro type?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
