private Kryo getKryoInstance ( ) { try { // check if ScalaKryoInstantiator is in class path (coming from Twitter's Chill library). // This will be true if Flink's Scala API is used. Class < ? > chillInstantiatorClazz = Class . forName ( "org.apache.flink.runtime.types.FlinkScalaKryoInstantiator" ) ; Object chillInstantiator = chillInstantiatorClazz . newInstance ( ) ; // obtain a Kryo instance through Twitter Chill Method m = chillInstantiatorClazz . getMethod ( "newKryo" ) ; return ( Kryo ) m . invoke ( chillInstantiator ) ; } catch ( ClassNotFoundException | InstantiationException | NoSuchMethodException | IllegalAccessException | InvocationTargetException e ) { LOG . warn ( "Falling back to default Kryo serializer because Chill serializer couldn't be found." , e ) ; Kryo . DefaultInstantiatorStrategy initStrategy = new Kryo . DefaultInstantiatorStrategy ( ) ; initStrategy . setFallbackInstantiatorStrategy ( new StdInstantiatorStrategy ( ) ) ; Kryo kryo = new Kryo ( ) ; kryo . setInstantiatorStrategy ( initStrategy ) ; return kryo ; } }
Returns the Chill Kryo Serializer which is implicitly added to the classpath via flink - runtime . Falls back to the default Kryo serializer if it can t be found .