Hi Timo, Please find the generated class (for the second schema) attached.
Regards, Petter On Wed, Apr 18, 2018 at 11:32 AM, Timo Walther <twal...@apache.org> wrote: > Hi Petter, > > could you share the source code of the class that Avro generates out of > this schema? > > Thank you. > > Regards, > Timo > > Am 18.04.18 um 11:00 schrieb Petter Arvidsson: > > Hello everyone, > > I am trying to figure out how to set up Flink with Avro for state > management (especially the content of snapshots) to enable state migrations > (e.g. adding a nullable fields to a class). So far, in version 1.4.2, I > tried to explicitly provide an instance of "new > AvroTypeInfo(Accumulator.getClass())" > where accumulator is a very simple Avro generated SpecificRecordBase of the > following schema: > > {"namespace": "io.relayr.flink", > "type": "record", > "name": "Accumulator", > "fields": [ > {"name": "accumulator", "type": "int"} > ] > } > > This successfully saves the state to the snapshot. When I then try to load > the snapshot with an updated schema (adding the nullable field) it fails. > Schema looks like this: > > {"namespace": "io.relayr.flink", > "type": "record", > "name": "Accumulator", > "fields": [ > {"name": "accumulator", "type": "int"}, > {"name": "newStuff", "type": ["int", "null"]} > ] > } > > When I try to restart the Job from the snapshot, I get the following > exception: > 2018-04-17 09:35:23,519 WARN org.apache.flink.api.common.ty > peutils.TypeSerializerSerializationUtil - Deserialization of serializer > errored; replacing with null. > java.io.IOException: Unloadable class for type serializer. > ... > Caused by: java.io.InvalidClassException: io.relayr.flink.Accumulator; > local class incompatible: stream classdesc serialVersionUID = > -3555733236161157838, local class serialVersionUID = 5291033088112484292 > > Which is true, Avro tools do generate a new serialization ID for the bean, > I just didn't expect it to be used and expected the Avro schema to be used > instead? Did anyone get this working? What am I getting wrong? > > Best regards, > Petter > > >
/** * Autogenerated by Avro * * DO NOT EDIT DIRECTLY */ package io.relayr.flink; import org.apache.avro.specific.SpecificData; import org.apache.avro.message.BinaryMessageEncoder; import org.apache.avro.message.BinaryMessageDecoder; import org.apache.avro.message.SchemaStore; @SuppressWarnings("all") @org.apache.avro.specific.AvroGenerated public class Accumulator extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord { private static final long serialVersionUID = 5291033088112484292L; public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Accumulator\",\"namespace\":\"io.relayr.flink\",\"fields\":[{\"name\":\"accumulator\",\"type\":\"int\"},{\"name\":\"newStuff\",\"type\":[\"int\",\"null\"]}]}"); public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; } private static SpecificData MODEL$ = new SpecificData(); private static final BinaryMessageEncoder<Accumulator> ENCODER = new BinaryMessageEncoder<Accumulator>(MODEL$, SCHEMA$); private static final BinaryMessageDecoder<Accumulator> DECODER = new BinaryMessageDecoder<Accumulator>(MODEL$, SCHEMA$); /** * Return the BinaryMessageDecoder instance used by this class. */ public static BinaryMessageDecoder<Accumulator> getDecoder() { return DECODER; } /** * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}. * @param resolver a {@link SchemaStore} used to find schemas by fingerprint */ public static BinaryMessageDecoder<Accumulator> createDecoder(SchemaStore resolver) { return new BinaryMessageDecoder<Accumulator>(MODEL$, SCHEMA$, resolver); } /** Serializes this Accumulator to a ByteBuffer. */ public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException { return ENCODER.encode(this); } /** Deserializes a Accumulator from a ByteBuffer. */ public static Accumulator fromByteBuffer( java.nio.ByteBuffer b) throws java.io.IOException { return DECODER.decode(b); } @Deprecated public int accumulator; @Deprecated public java.lang.Integer newStuff; /** * Default constructor. Note that this does not initialize fields * to their default values from the schema. If that is desired then * one should use <code>newBuilder()</code>. */ public Accumulator() {} /** * All-args constructor. * @param accumulator The new value for accumulator * @param newStuff The new value for newStuff */ public Accumulator(java.lang.Integer accumulator, java.lang.Integer newStuff) { this.accumulator = accumulator; this.newStuff = newStuff; } public org.apache.avro.Schema getSchema() { return SCHEMA$; } // Used by DatumWriter. Applications should not call. public java.lang.Object get(int field$) { switch (field$) { case 0: return accumulator; case 1: return newStuff; default: throw new org.apache.avro.AvroRuntimeException("Bad index"); } } // Used by DatumReader. Applications should not call. @SuppressWarnings(value="unchecked") public void put(int field$, java.lang.Object value$) { switch (field$) { case 0: accumulator = (java.lang.Integer)value$; break; case 1: newStuff = (java.lang.Integer)value$; break; default: throw new org.apache.avro.AvroRuntimeException("Bad index"); } } /** * Gets the value of the 'accumulator' field. * @return The value of the 'accumulator' field. */ public java.lang.Integer getAccumulator() { return accumulator; } /** * Sets the value of the 'accumulator' field. * @param value the value to set. */ public void setAccumulator(java.lang.Integer value) { this.accumulator = value; } /** * Gets the value of the 'newStuff' field. * @return The value of the 'newStuff' field. */ public java.lang.Integer getNewStuff() { return newStuff; } /** * Sets the value of the 'newStuff' field. * @param value the value to set. */ public void setNewStuff(java.lang.Integer value) { this.newStuff = value; } /** * Creates a new Accumulator RecordBuilder. * @return A new Accumulator RecordBuilder */ public static io.relayr.flink.Accumulator.Builder newBuilder() { return new io.relayr.flink.Accumulator.Builder(); } /** * Creates a new Accumulator RecordBuilder by copying an existing Builder. * @param other The existing builder to copy. * @return A new Accumulator RecordBuilder */ public static io.relayr.flink.Accumulator.Builder newBuilder(io.relayr.flink.Accumulator.Builder other) { return new io.relayr.flink.Accumulator.Builder(other); } /** * Creates a new Accumulator RecordBuilder by copying an existing Accumulator instance. * @param other The existing instance to copy. * @return A new Accumulator RecordBuilder */ public static io.relayr.flink.Accumulator.Builder newBuilder(io.relayr.flink.Accumulator other) { return new io.relayr.flink.Accumulator.Builder(other); } /** * RecordBuilder for Accumulator instances. */ public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Accumulator> implements org.apache.avro.data.RecordBuilder<Accumulator> { private int accumulator; private java.lang.Integer newStuff; /** Creates a new Builder */ private Builder() { super(SCHEMA$); } /** * Creates a Builder by copying an existing Builder. * @param other The existing Builder to copy. */ private Builder(io.relayr.flink.Accumulator.Builder other) { super(other); if (isValidValue(fields()[0], other.accumulator)) { this.accumulator = data().deepCopy(fields()[0].schema(), other.accumulator); fieldSetFlags()[0] = true; } if (isValidValue(fields()[1], other.newStuff)) { this.newStuff = data().deepCopy(fields()[1].schema(), other.newStuff); fieldSetFlags()[1] = true; } } /** * Creates a Builder by copying an existing Accumulator instance * @param other The existing instance to copy. */ private Builder(io.relayr.flink.Accumulator other) { super(SCHEMA$); if (isValidValue(fields()[0], other.accumulator)) { this.accumulator = data().deepCopy(fields()[0].schema(), other.accumulator); fieldSetFlags()[0] = true; } if (isValidValue(fields()[1], other.newStuff)) { this.newStuff = data().deepCopy(fields()[1].schema(), other.newStuff); fieldSetFlags()[1] = true; } } /** * Gets the value of the 'accumulator' field. * @return The value. */ public java.lang.Integer getAccumulator() { return accumulator; } /** * Sets the value of the 'accumulator' field. * @param value The value of 'accumulator'. * @return This builder. */ public io.relayr.flink.Accumulator.Builder setAccumulator(int value) { validate(fields()[0], value); this.accumulator = value; fieldSetFlags()[0] = true; return this; } /** * Checks whether the 'accumulator' field has been set. * @return True if the 'accumulator' field has been set, false otherwise. */ public boolean hasAccumulator() { return fieldSetFlags()[0]; } /** * Clears the value of the 'accumulator' field. * @return This builder. */ public io.relayr.flink.Accumulator.Builder clearAccumulator() { fieldSetFlags()[0] = false; return this; } /** * Gets the value of the 'newStuff' field. * @return The value. */ public java.lang.Integer getNewStuff() { return newStuff; } /** * Sets the value of the 'newStuff' field. * @param value The value of 'newStuff'. * @return This builder. */ public io.relayr.flink.Accumulator.Builder setNewStuff(java.lang.Integer value) { validate(fields()[1], value); this.newStuff = value; fieldSetFlags()[1] = true; return this; } /** * Checks whether the 'newStuff' field has been set. * @return True if the 'newStuff' field has been set, false otherwise. */ public boolean hasNewStuff() { return fieldSetFlags()[1]; } /** * Clears the value of the 'newStuff' field. * @return This builder. */ public io.relayr.flink.Accumulator.Builder clearNewStuff() { newStuff = null; fieldSetFlags()[1] = false; return this; } @Override @SuppressWarnings("unchecked") public Accumulator build() { try { Accumulator record = new Accumulator(); record.accumulator = fieldSetFlags()[0] ? this.accumulator : (java.lang.Integer) defaultValue(fields()[0]); record.newStuff = fieldSetFlags()[1] ? this.newStuff : (java.lang.Integer) defaultValue(fields()[1]); return record; } catch (java.lang.Exception e) { throw new org.apache.avro.AvroRuntimeException(e); } } } @SuppressWarnings("unchecked") private static final org.apache.avro.io.DatumWriter<Accumulator> WRITER$ = (org.apache.avro.io.DatumWriter<Accumulator>)MODEL$.createDatumWriter(SCHEMA$); @Override public void writeExternal(java.io.ObjectOutput out) throws java.io.IOException { WRITER$.write(this, SpecificData.getEncoder(out)); } @SuppressWarnings("unchecked") private static final org.apache.avro.io.DatumReader<Accumulator> READER$ = (org.apache.avro.io.DatumReader<Accumulator>)MODEL$.createDatumReader(SCHEMA$); @Override public void readExternal(java.io.ObjectInput in) throws java.io.IOException { READER$.read(this, SpecificData.getDecoder(in)); } }