Hi. Thanks for the tip. It seems to work...
Greets. > Am 18.08.2015 um 13:56 schrieb Stephan Ewen <se...@apache.org>: > > Yep, that is a valid bug! > State is apparently not resolved with the correct classloader. > > As a workaround, you can checkpoint byte arrays and serialize/deserialize the > state into byte arrays yourself. You can use the apache commons > SerializationUtil class, or Flinks InstantiationUtil class for that. > > You can get the ClassLoader for the user code (needed for deserialization) > via "getRuntimeContext().getUserCodeClassLoader()". > > Let us know if that workaround works. We'll try to get a fix for that out > very soon! > > Greetings, > Stephan > > > >> On Tue, Aug 18, 2015 at 12:23 PM, Robert Metzger <rmetz...@apache.org> wrote: >> Java's HashMap is serializable. >> If it is only the map, you can just use the HashMap<> as the state. >> >> If you have more data, you can use TupleX, for example: >> >> Tuple2<HashMap<Integer, String>, Long>(myMap, myLong); >> >> >>> On Tue, Aug 18, 2015 at 12:21 PM, Rico Bergmann <i...@ricobergmann.de> >>> wrote: >>> Hi! >>> >>> Using TupleX is not possible since the state is very big (a Hashtable). >>> >>> How would I have to do serialization into a byte array? >>> >>> Greets. Rico. >>> >>> >>> >>>> Am 18.08.2015 um 11:44 schrieb Robert Metzger <rmetz...@apache.org>: >>>> >>>> Hi Rico, >>>> >>>> I'm pretty sure that this is a valid bug you've found, since this case is >>>> not yet tested (afaik). >>>> We'll fix the issue asap, until then, are you able to encapsulate your >>>> state in something that is available in Flink, for example a TupleX or >>>> just serialize it yourself into a byte[] ? >>>> >>>>> On Tue, Aug 18, 2015 at 11:32 AM, Rico Bergmann <i...@ricobergmann.de> >>>>> wrote: >>>>> Hi! >>>>> Is it possible to use your own class? >>>>> I'm using the file state handler at the Jobmanager and implemented the >>>>> Checkpointed interface. >>>>> >>>>> I tried this and got an exception: >>>>> >>>>> Error: java.lang.RuntimeException: Failed to deserialize state handle and >>>>> setup initial operator state. >>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:544) >>>>> at java.lang.Thread.run(Thread.java:745) >>>>> Caused by: java.lang.ClassNotFoundException: >>>>> com.ottogroup.bi.searchlab.searchsessionizer.OperatorState >>>>>> at java.net.URLClassLoader.findClass(URLClassLoader.java:381) >>>>>> at java.lang.ClassLoader.loadClass(ClassLoader.java:424) >>>>>> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331) >>>>>> at java.lang.ClassLoader.loadClass(ClassLoader.java:357) >>>>>> at java.lang.Class.forName0(Native Method) >>>>>> at java.lang.Class.forName(Class.java:348) >>>>>> at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:626) >>>>>> at >>>>>> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1613) >>>>>> at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1518) >>>>>> at >>>>>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1774) >>>>>> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) >>>>>> at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371) >>>>>> at >>>>>> org.apache.flink.runtime.state.ByteStreamStateHandle.getState(ByteStreamStateHandle.java:63) >>>>>> at >>>>>> org.apache.flink.runtime.state.ByteStreamStateHandle.getState(ByteStreamStateHandle.java:33) >>>>>> at >>>>>> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.restoreInitialState(AbstractUdfStreamOperator.java:83) >>>>>> at >>>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.setInitialState(StreamTask.java:276) >>>>>> at >>>>>> org.apache.flink.runtime.state.StateUtils.setOperatorState(StateUtils.java:51) >>>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:541) >
