gyfora commented on code in PR #25896:
URL: https://github.com/apache/flink/pull/25896#discussion_r1975030496
##########
flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/utils/AvroKryoSerializerUtils.java:
##########
@@ -106,11 +105,55 @@ public void write(Kryo kryo, Output output, Schema
object) {
}
@Override
- public Schema read(Kryo kryo, Input input, Class<Schema> type) {
+ public Schema read(Kryo kryo, Input input, Class<? extends Schema>
type) {
String schemaAsString = input.readString();
// the parser seems to be stateful, to we need a new one for every
type.
Schema.Parser sParser = new Schema.Parser();
return sParser.parse(schemaAsString);
}
}
+
+ public static class AvroGenericDataArraySerializer
+ extends CollectionSerializer<GenericData.Array> implements
Serializable {
+ @Override
+ public void write(Kryo kryo, Output output, GenericData.Array
collection) {
+ String schemaAsString = collection.getSchema().toString(false);
+ output.writeString(schemaAsString);
+
+ int length = collection.size();
+ output.writeVarInt(length + 1, true);
+
+ // Efficiency could be improved for cases where all elements are
the same type.
+ for (Object element : collection) {
+ kryo.writeClassAndObject(output, element);
+ }
+ }
+
+ @Override
+ public GenericData.Array read(
+ Kryo kryo, Input input, Class<? extends GenericData.Array>
type) {
+ String schemaAsString = input.readString();
+ Schema.Parser schemaParser = new Schema.Parser();
+ Schema schema = schemaParser.parse(schemaAsString);
+
+ int lengthPlusOne = input.readVarInt(true);
Review Comment:
Why are we writing/reading length + 1 instead of length?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
