[refacto] TypeInformation.createSerializer(ExecutionConfig) should continue to work

Author: arnaud-daroussin

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/CaseClassTypeInfo.scala

@@ -49,7 +49,7 @@ class CaseClassTypeInfo[T <: Product](
   override def createSerializer(config: SerializerConfig): TypeSerializer[T] = ser.duplicate()
 
   // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
-  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = null
+  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = ser.duplicate()
 
   private val REGEX_INT_FIELD: String              = "[0-9]+"
   private val REGEX_STR_FIELD: String              = "[\\p{L}_\\$][\\p{L}\\p{Digit}_\\$]*"

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/CollectionTypeInformation.scala

@@ -12,7 +12,7 @@ case class CollectionTypeInformation[T: ClassTag](serializer: TypeSerializer[T])
 
   override def createSerializer(config: SerializerConfig): TypeSerializer[T] = serializer.duplicate()
   // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
-  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = null
+  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = serializer.duplicate()
 
   override def isBasicType: Boolean   = false
   override def isTupleType: Boolean   = false

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/CoproductTypeInformation.scala

@@ -9,7 +9,7 @@ case class CoproductTypeInformation[T](c: Class[T], ser: TypeSerializer[T]) exte
 
   override def createSerializer(config: SerializerConfig): TypeSerializer[T] = ser.duplicate()
   // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
-  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = null
+  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = ser.duplicate()
 
   override def isBasicType: Boolean   = false
   override def isTupleType: Boolean   = false

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/EitherTypeInfo.scala

@@ -70,7 +70,7 @@ class EitherTypeInfo[A, B, T <: Either[A, B]](
   }
 
   // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
-  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = null
+  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = createSerializer(config.getSerializerConfig)
 
   override def equals(obj: Any): Boolean = {
     obj match {

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/MappedTypeInformation.scala

@@ -15,7 +15,7 @@ case class MappedTypeInformation[A: ClassTag, B](mapper: TypeMapper[A, B], neste
   override def createSerializer(config: SerializerConfig): TypeSerializer[A] =
     new MappedSerializer(mapper, nested.createSerializer(config))
   // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
-  def createSerializer(config: ExecutionConfig): TypeSerializer[A] = null
+  def createSerializer(config: ExecutionConfig): TypeSerializer[A] = createSerializer(config.getSerializerConfig)
 
   override def isKeyType: Boolean   = nested.isKeyType
   override def getTotalFields: Int  = nested.getTotalFields

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/OptionTypeInfo.scala

@@ -74,7 +74,7 @@ class OptionTypeInfo[A, T <: Option[A]](private val elemTypeInfo: TypeInformatio
     }
 
   // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
-  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = null
+  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = createSerializer(config.getSerializerConfig)
 
   override def toString = s"Option[$elemTypeInfo]"
 

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/SimpleTypeInformation.scala

@@ -11,7 +11,7 @@ abstract class SimpleTypeInformation[T: ClassTag: TypeSerializer] extends TypeIn
 
   override def createSerializer(config: SerializerConfig): TypeSerializer[T] = implicitly[TypeSerializer[T]].duplicate()
   // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
-  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = null
+  def createSerializer(config: ExecutionConfig): TypeSerializer[T] = implicitly[TypeSerializer[T]].duplicate()
 
   override def isBasicType: Boolean   = false
   override def isTupleType: Boolean   = false

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/typeinfo/UnitTypeInformation.scala

@@ -10,7 +10,7 @@ class UnitTypeInformation extends TypeInformation[Unit] {
 
   override def createSerializer(config: SerializerConfig): TypeSerializer[Unit] = new UnitSerializer()
   // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
-  def createSerializer(config: ExecutionConfig): TypeSerializer[Unit] = null
+  def createSerializer(config: ExecutionConfig): TypeSerializer[Unit] = new UnitSerializer()
 
   override def isKeyType: Boolean          = false
   override def getTotalFields: Int         = 0