feat(#286): fix CI "No ClassTag available for Nothing" error

Author: arnaud-daroussin

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

@@ -33,7 +33,6 @@ import java.math.{BigInteger => JBigInteger}
 import java.math.{BigDecimal => JBigDecimal}
 import java.time._
 import java.util.UUID
-import scala.{Nothing => Nothing}
 import scala.collection.immutable.{SortedSet, TreeSet}
 import scala.concurrent.duration.{Duration, FiniteDuration, TimeUnit}
 import scala.reflect.{ClassTag, classTag}
@@ -116,9 +115,7 @@ trait serializers extends LowPrioImplicits {
   implicit lazy val finiteDurationInfo: TypeInformation[FiniteDuration] = SimpleTypeInfo(2, 2, keyType = true)
   implicit lazy val flinkNothingInfo: TypeInformation[FlinkNothing]     =
     SimpleTypeInfo(0)(classTag[FlinkNothing], new NothingSerializer().asInstanceOf[TypeSerializer[FlinkNothing]])
-  implicit lazy val scalaNothingInfo: TypeInformation[Nothing] =
-    // At runtime scala.Nothing is scala.runtime.Nothing$ and extends Throwable
-    SimpleTypeInfo[Nothing](6, 11)(classTag[Nothing], new NothingSerializer().asInstanceOf[TypeSerializer[Nothing]])
+  implicit lazy val scalaNothingInfo: TypeInformation[scala.Nothing]  = ScalaNothingTypeInfo
   implicit lazy val timeUnitInfo: TypeInformation[TimeUnit]           = new EnumTypeInfo(classOf[TimeUnit])
   implicit lazy val uuidMapper: TypeMapper[UUID, Array[Byte]]         = new UuidMapper()
   implicit lazy val uuidInfo: TypeInformation[UUID]                   = mappedTypeInfo[UUID, Array[Byte]]

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

@@ -0,0 +1,28 @@
+package org.apache.flinkx.api.typeinfo
+
+import org.apache.flink.api.common.ExecutionConfig
+import org.apache.flink.api.common.serialization.SerializerConfig
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.common.typeutils.TypeSerializer
+import org.apache.flinkx.api.serializer.NothingSerializer
+
+object ScalaNothingTypeInfo extends TypeInformation[Nothing] {
+
+  override def createSerializer(config: SerializerConfig): TypeSerializer[Nothing] =
+    new NothingSerializer().asInstanceOf[TypeSerializer[Nothing]]
+  // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
+  def createSerializer(config: ExecutionConfig): TypeSerializer[Nothing] =
+    new NothingSerializer().asInstanceOf[TypeSerializer[Nothing]]
+
+  override def isKeyType: Boolean   = false
+  override def getTotalFields: Int  = 11 // At runtime scala.Nothing is scala.runtime.Nothing$ and extends Throwable
+  override def isTupleType: Boolean = false
+  override def canEqual(obj: Any): Boolean  = obj.isInstanceOf[ScalaNothingTypeInfo.type]
+  override def getTypeClass: Class[Nothing] = classOf[Nothing]
+  override def getArity: Int             = 6 // At runtime scala.Nothing is scala.runtime.Nothing$ and extends Throwable
+  override def isBasicType: Boolean      = false
+  override def toString: String          = "ScalaNothingTypeInfo()"
+  override def equals(obj: Any): Boolean = this.eq(obj.asInstanceOf[AnyRef])
+  override def hashCode(): Int           = 0
+
+}