This is an automated email from the ASF dual-hosted git repository. aljoscha pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5f688b4c6fb469f8c7447d670e6ca33643424a2f Author: Aljoscha Krettek <[email protected]> AuthorDate: Thu Oct 18 15:34:52 2018 +0200 [FLINK-7811] Add ScalaKryoInstantiator from Chill 0.7.4 and update to 0.7.6 We have to do this in order to be able to update our Chill dependency without changing the the Kryo serializers that are registered by default. The problem is that snapshots of our Kryo serializer only contain the user-registered serializers, not the serializers that are registered by default by Chill. If we had that, we could probably get by without this change. The reason we have to update is that there is no Chill 0.7.4 dependency for Scala 2.12. --- .../typeutils/runtime/kryo/KryoSerializer.java | 3 +- flink-formats/flink-avro/pom.xml | 9 + .../runtime/types/FlinkChillPackageRegistrar.java | 63 +++++++ .../runtime/types/PriorityQueueSerializer.java | 93 ++++++++++ .../runtime/types/FlinkScalaKryoInstantiator.scala | 194 +++++++++++++++++++++ .../types/JavaIterableWrapperSerializer.scala | 75 ++++++++ flink-scala/pom.xml | 8 + pom.xml | 2 +- 8 files changed, 445 insertions(+), 2 deletions(-) diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java index 5ac914e..af36655 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java @@ -401,7 +401,8 @@ public class KryoSerializer<T> extends TypeSerializer<T> { 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("com.twitter.chill.ScalaKryoInstantiator"); + Class<?> chillInstantiatorClazz = + Class.forName("org.apache.flink.runtime.types.FlinkScalaKryoInstantiator"); Object chillInstantiator = chillInstantiatorClazz.newInstance(); // obtain a Kryo instance through Twitter Chill diff --git a/flink-formats/flink-avro/pom.xml b/flink-formats/flink-avro/pom.xml index 4f62a7e..421c91e 100644 --- a/flink-formats/flink-avro/pom.xml +++ b/flink-formats/flink-avro/pom.xml @@ -120,6 +120,15 @@ under the License. <scope>test</scope> <type>test-jar</type> </dependency> + + <!-- We need this for the patched FlinkScalaKryoInstantiator --> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-runtime_${scala.binary.version}</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + </dependencies> <profiles> diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/types/FlinkChillPackageRegistrar.java b/flink-runtime/src/main/java/org/apache/flink/runtime/types/FlinkChillPackageRegistrar.java new file mode 100644 index 0000000..0e470a1 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/types/FlinkChillPackageRegistrar.java @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.types; + +import com.twitter.chill.IKryoRegistrar; +import com.twitter.chill.java.ArraysAsListSerializer; +import com.twitter.chill.java.BitSetSerializer; +import com.twitter.chill.java.InetSocketAddressSerializer; +import com.twitter.chill.java.IterableRegistrar; +import com.twitter.chill.java.LocaleSerializer; +import com.twitter.chill.java.RegexSerializer; +import com.twitter.chill.java.SimpleDateFormatSerializer; +import com.twitter.chill.java.SqlDateSerializer; +import com.twitter.chill.java.SqlTimeSerializer; +import com.twitter.chill.java.TimestampSerializer; +import com.twitter.chill.java.URISerializer; +import com.twitter.chill.java.UUIDSerializer; + +/* +This code is copied as is from Twitter Chill 0.7.4 because we need to user a newer chill version +but want to ensure that the serializers that are registered by default stay the same. + +The only changes to the code are those that are required to make it compile and pass checkstyle +checks in our code base. + */ + +/** + * Creates a registrar for all the serializers in the chill.java package. + */ +public class FlinkChillPackageRegistrar { + + public static IKryoRegistrar all() { + return new IterableRegistrar( + ArraysAsListSerializer.registrar(), + BitSetSerializer.registrar(), + PriorityQueueSerializer.registrar(), + RegexSerializer.registrar(), + SqlDateSerializer.registrar(), + SqlTimeSerializer.registrar(), + TimestampSerializer.registrar(), + URISerializer.registrar(), + InetSocketAddressSerializer.registrar(), + UUIDSerializer.registrar(), + LocaleSerializer.registrar(), + SimpleDateFormatSerializer.registrar()); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/types/PriorityQueueSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/types/PriorityQueueSerializer.java new file mode 100644 index 0000000..c9bb5f9 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/types/PriorityQueueSerializer.java @@ -0,0 +1,93 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.types; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Serializer; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; +import com.twitter.chill.IKryoRegistrar; +import com.twitter.chill.SingleRegistrar; + +import java.lang.reflect.Field; +import java.util.Comparator; +import java.util.PriorityQueue; + +/* +This code is copied as is from Twitter Chill 0.7.4 because we need to user a newer chill version +but want to ensure that the serializers that are registered by default stay the same. + +The only changes to the code are those that are required to make it compile and pass checkstyle +checks in our code base. + */ + +class PriorityQueueSerializer extends Serializer<PriorityQueue<?>> { + private Field compField; + + public static IKryoRegistrar registrar() { + return new SingleRegistrar(PriorityQueue.class, new PriorityQueueSerializer()); + } + + public PriorityQueueSerializer() { + try { + compField = PriorityQueue.class.getDeclaredField("comparator"); + compField.setAccessible(true); + } + catch (Exception e) { + throw new RuntimeException(e); + } + } + + public Comparator<?> getComparator(PriorityQueue<?> q) { + try { + return (Comparator<?>) compField.get(q); + } + catch (Exception e) { + throw new RuntimeException(e); + } + } + + public void write(Kryo k, Output o, PriorityQueue<?> q) { + k.writeClassAndObject(o, getComparator(q)); + o.writeInt(q.size(), true); + for (Object a : q) { + k.writeClassAndObject(o, a); + o.flush(); + } + } + + public PriorityQueue<?> read(Kryo k, Input i, Class<PriorityQueue<?>> c) { + Comparator<Object> comp = (Comparator<Object>) k.readClassAndObject(i); + int sz = i.readInt(true); + // can't create with size 0: + PriorityQueue<Object> result; + if (sz == 0) { + result = new PriorityQueue<Object>(1, comp); + } + else { + result = new PriorityQueue<Object>(sz, comp); + } + int idx = 0; + while (idx < sz) { + result.add(k.readClassAndObject(i)); + idx += 1; + } + return result; + } +} diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/types/FlinkScalaKryoInstantiator.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/types/FlinkScalaKryoInstantiator.scala new file mode 100644 index 0000000..b98f0e5 --- /dev/null +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/types/FlinkScalaKryoInstantiator.scala @@ -0,0 +1,194 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.types + +import scala.collection.immutable.{BitSet, HashMap, HashSet, ListMap, ListSet, NumericRange, Queue, Range, SortedMap, SortedSet} +import scala.collection.mutable.{Buffer, ListBuffer, WrappedArray, BitSet => MBitSet, HashMap => MHashMap, HashSet => MHashSet, Map => MMap, Queue => MQueue, Set => MSet} +import scala.util.matching.Regex +import _root_.java.io.Serializable + +import com.twitter.chill._ + + +import scala.collection.JavaConverters._ + +/* +This code is copied as is from Twitter Chill 0.7.4 because we need to user a newer chill version +but want to ensure that the serializers that are registered by default stay the same. + +The only changes to the code are those that are required to make it compile and pass checkstyle +checks in our code base. + */ + +/** + * This class has a no-arg constructor, suitable for use with reflection instantiation + * It has no registered serializers, just the standard Kryo configured for Kryo. + */ +class EmptyFlinkScalaKryoInstantiator extends KryoInstantiator { + override def newKryo = { + val k = new KryoBase + k.setRegistrationRequired(false) + k.setInstantiatorStrategy(new org.objenesis.strategy.StdInstantiatorStrategy) + + // Handle cases where we may have an odd classloader setup like with libjars + // for hadoop + val classLoader = Thread.currentThread.getContextClassLoader + k.setClassLoader(classLoader) + + k + } +} + +object FlinkScalaKryoInstantiator extends Serializable { + private val mutex = new AnyRef with Serializable // some serializable object + @transient private var kpool: KryoPool = null + + /** + * Return a KryoPool that uses the FlinkScalaKryoInstantiator + */ + def defaultPool: KryoPool = mutex.synchronized { + if (null == kpool) { + kpool = KryoPool.withByteArrayOutputStream(guessThreads, new FlinkScalaKryoInstantiator) + } + kpool + } + + private def guessThreads: Int = { + val cores = Runtime.getRuntime.availableProcessors + val GUESS_THREADS_PER_CORE = 4 + GUESS_THREADS_PER_CORE * cores + } +} + +/** Makes an empty instantiator then registers everything */ +class FlinkScalaKryoInstantiator extends EmptyFlinkScalaKryoInstantiator { + override def newKryo = { + val k = super.newKryo + val reg = new AllScalaRegistrar + reg(k) + k + } +} + +class ScalaCollectionsRegistrar extends IKryoRegistrar { + def apply(newK: Kryo) { + // for binary compat this is here, but could be moved to RichKryo + def useField[T](cls: Class[T]) { + val fs = new com.esotericsoftware.kryo.serializers.FieldSerializer(newK, cls) + fs.setIgnoreSyntheticFields(false) // scala generates a lot of these attributes + newK.register(cls, fs) + } + // The wrappers are private classes: + useField(List(1, 2, 3).asJava.getClass) + useField(List(1, 2, 3).iterator.asJava.getClass) + useField(Map(1 -> 2, 4 -> 3).asJava.getClass) + useField(new _root_.java.util.ArrayList().asScala.getClass) + useField(new _root_.java.util.HashMap().asScala.getClass) + + /* + * Note that subclass-based use: addDefaultSerializers, else: register + * You should go from MOST specific, to least to specific when using + * default serializers. The FIRST one found is the one used + */ + newK + // wrapper array is abstract + .forSubclass[WrappedArray[Any]](new WrappedArraySerializer[Any]) + .forSubclass[BitSet](new BitSetSerializer) + .forSubclass[SortedSet[Any]](new SortedSetSerializer) + .forClass[Some[Any]](new SomeSerializer[Any]) + .forClass[Left[Any, Any]](new LeftSerializer[Any, Any]) + .forClass[Right[Any, Any]](new RightSerializer[Any, Any]) + .forTraversableSubclass(Queue.empty[Any]) + // List is a sealed class, so there are only two subclasses: + .forTraversableSubclass(List.empty[Any]) + // Add ListBuffer subclass before Buffer to prevent the more general case taking precedence + .forTraversableSubclass(ListBuffer.empty[Any], isImmutable = false) + // add mutable Buffer before Vector, otherwise Vector is used + .forTraversableSubclass(Buffer.empty[Any], isImmutable = false) + // Vector is a final class + .forTraversableClass(Vector.empty[Any]) + .forTraversableSubclass(ListSet.empty[Any]) + // specifically register small sets since Scala represents them differently + .forConcreteTraversableClass(Set[Any]('a)) + .forConcreteTraversableClass(Set[Any]('a, 'b)) + .forConcreteTraversableClass(Set[Any]('a, 'b, 'c)) + .forConcreteTraversableClass(Set[Any]('a, 'b, 'c, 'd)) + // default set implementation + .forConcreteTraversableClass(HashSet[Any]('a, 'b, 'c, 'd, 'e)) + // specifically register small maps since Scala represents them differently + .forConcreteTraversableClass(Map[Any, Any]('a -> 'a)) + .forConcreteTraversableClass(Map[Any, Any]('a -> 'a, 'b -> 'b)) + .forConcreteTraversableClass(Map[Any, Any]('a -> 'a, 'b -> 'b, 'c -> 'c)) + .forConcreteTraversableClass(Map[Any, Any]('a -> 'a, 'b -> 'b, 'c -> 'c, 'd -> 'd)) + // default map implementation + .forConcreteTraversableClass( + HashMap[Any, Any]('a -> 'a, 'b -> 'b, 'c -> 'c, 'd -> 'd, 'e -> 'e)) + // The normal fields serializer works for ranges + .registerClasses(Seq(classOf[Range.Inclusive], + classOf[NumericRange.Inclusive[_]], + classOf[NumericRange.Exclusive[_]])) + // Add some maps + .forSubclass[SortedMap[Any, Any]](new SortedMapSerializer) + .forTraversableSubclass(ListMap.empty[Any, Any]) + .forTraversableSubclass(HashMap.empty[Any, Any]) + // The above ListMap/HashMap must appear before this: + .forTraversableSubclass(Map.empty[Any, Any]) + // here are the mutable ones: + .forTraversableClass(MBitSet.empty, isImmutable = false) + .forTraversableClass(MHashMap.empty[Any, Any], isImmutable = false) + .forTraversableClass(MHashSet.empty[Any], isImmutable = false) + .forTraversableSubclass(MQueue.empty[Any], isImmutable = false) + .forTraversableSubclass(MMap.empty[Any, Any], isImmutable = false) + .forTraversableSubclass(MSet.empty[Any], isImmutable = false) + } +} + +class JavaWrapperCollectionRegistrar extends IKryoRegistrar { + def apply(newK: Kryo) { + newK.register(JavaIterableWrapperSerializer.wrapperClass, new JavaIterableWrapperSerializer) + } +} + +/** Registers all the scala (and java) serializers we have */ +class AllScalaRegistrar extends IKryoRegistrar { + def apply(k: Kryo) { + val col = new ScalaCollectionsRegistrar + col(k) + + val jcol = new JavaWrapperCollectionRegistrar + jcol(k) + + // Register all 22 tuple serializers and specialized serializers + ScalaTupleSerialization.register(k) + k.forClass[Symbol](new KSerializer[Symbol] { + override def isImmutable = true + def write(k: Kryo, out: Output, obj: Symbol) { out.writeString(obj.name) } + def read(k: Kryo, in: Input, cls: Class[Symbol]) = Symbol(in.readString) + }) + .forSubclass[Regex](new RegexSerializer) + .forClass[ClassManifest[Any]](new ClassManifestSerializer[Any]) + .forSubclass[Manifest[Any]](new ManifestSerializer[Any]) + .forSubclass[scala.Enumeration#Value](new EnumerationSerializer) + + // use the singleton serializer for boxed Unit + val boxedUnit = scala.Unit.box(()) + k.register(boxedUnit.getClass, new SingletonSerializer(boxedUnit)) + FlinkChillPackageRegistrar.all()(k) + } +} diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/types/JavaIterableWrapperSerializer.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/types/JavaIterableWrapperSerializer.scala new file mode 100644 index 0000000..e46e4ce --- /dev/null +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/types/JavaIterableWrapperSerializer.scala @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.types + +import _root_.java.lang.{Iterable => JIterable} +import com.twitter.chill.{Input, KSerializer, Kryo, Output} + +/* +This code is copied as is from Twitter Chill 0.7.4 because we need to user a newer chill version +but want to ensure that the serializers that are registered by default stay the same. + +The only changes to the code are those that are required to make it compile and pass checkstyle +checks in our code base. + */ + +/** + * A Kryo serializer for serializing results returned by asJavaIterable. + * + * The underlying object is scala.collection.convert.Wrappers$IterableWrapper. + * Kryo deserializes this into an AbstractCollection, which unfortunately doesn't work. + * + * Ported from Apache Spark's KryoSerializer.scala. + */ +private class JavaIterableWrapperSerializer extends KSerializer[JIterable[_]] { + + import JavaIterableWrapperSerializer._ + + override def write(kryo: Kryo, out: Output, obj: JIterable[_]): Unit = { + // If the object is the wrapper, simply serialize the underlying Scala Iterable object. + // Otherwise, serialize the object itself. + if (obj.getClass == wrapperClass && underlyingMethodOpt.isDefined) { + kryo.writeClassAndObject(out, underlyingMethodOpt.get.invoke(obj)) + } else { + kryo.writeClassAndObject(out, obj) + } + } + + override def read(kryo: Kryo, in: Input, clz: Class[JIterable[_]]): JIterable[_] = { + kryo.readClassAndObject(in) match { + case scalaIterable: Iterable[_] => + scala.collection.JavaConversions.asJavaIterable(scalaIterable) + case javaIterable: JIterable[_] => + javaIterable + } + } +} + +private object JavaIterableWrapperSerializer { + // The class returned by asJavaIterable (scala.collection.convert.Wrappers$IterableWrapper). + val wrapperClass = scala.collection.JavaConversions.asJavaIterable(Seq(1)).getClass + + // Get the underlying method so we can use it to get the Scala collection for serialization. + private val underlyingMethodOpt = { + try Some(wrapperClass.getDeclaredMethod("underlying")) catch { + case e: Exception => + None + } + } +} diff --git a/flink-scala/pom.xml b/flink-scala/pom.xml index f5f6132..e7fc6e1 100644 --- a/flink-scala/pom.xml +++ b/flink-scala/pom.xml @@ -90,6 +90,14 @@ under the License. <scope>test</scope> </dependency> + <!-- We need this for the patched FlinkScalaKryoInstantiator --> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-runtime_${scala.binary.version}</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-core</artifactId> diff --git a/pom.xml b/pom.xml index 324a4c1..4b75c10 100644 --- a/pom.xml +++ b/pom.xml @@ -115,7 +115,7 @@ under the License. <!-- Default scala versions, may be overwritten by build profiles --> <scala.version>2.11.12</scala.version> <scala.binary.version>2.11</scala.binary.version> - <chill.version>0.7.4</chill.version> + <chill.version>0.7.6</chill.version> <zookeeper.version>3.4.10</zookeeper.version> <curator.version>2.12.0</curator.version> <jackson.version>2.7.9</jackson.version>
