Scala 2: Recursive type info is not supported: fail-fast with a comprehensive message instead (#280)

Author: arnaud-daroussin

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

@@ -4,8 +4,9 @@ import magnolia1.{CaseClass, Magnolia, SealedTrait}
 import org.apache.flink.api.common.serialization.SerializerConfig
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.typeutils.runtime.NullableSerializer
+import org.apache.flink.util.FlinkRuntimeException
 import org.apache.flinkx.api.serializer.{CaseClassSerializer, CoproductSerializer, ScalaCaseObjectSerializer, nullable}
-import org.apache.flinkx.api.typeinfo.{CaseClassTypeInfo, CoproductTypeInformation}
+import org.apache.flinkx.api.typeinfo.{CaseClassTypeInfo, CoproductTypeInformation, MarkerTypeInfo}
 import org.apache.flinkx.api.util.ClassUtil.isCaseClassImmutable
 
 import scala.collection.mutable
@@ -26,8 +27,11 @@ private[api] trait LowPrioImplicits {
   ): TypeInformation[T] = {
     val cacheKey = typeName[T]
     cache.get(cacheKey) match {
+      case Some(MarkerTypeInfo) =>
+        throw new FlinkRuntimeException(s"Unsupported: recursivity detected in '$cacheKey'.")
       case Some(cached) => cached.asInstanceOf[TypeInformation[T]]
       case None         =>
+        cache.put(cacheKey, MarkerTypeInfo)
         val clazz      = classTag[T].runtimeClass.asInstanceOf[Class[T]]
         val serializer = if (typeOf[T].typeSymbol.isModuleClass) {
           new ScalaCaseObjectSerializer[T](clazz)

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

@@ -0,0 +1,24 @@
+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
+
+case object MarkerTypeInfo extends TypeInformation[Nothing] {
+
+  override def createSerializer(config: SerializerConfig): TypeSerializer[Nothing] = null
+  // override modifier removed to satisfy both implementation requirement of Flink 1.x and removal in 2.x
+  def createSerializer(config: ExecutionConfig): TypeSerializer[Nothing] = null
+
+  override def isBasicType: Boolean         = false
+  override def isTupleType: Boolean         = false
+  override def isKeyType: Boolean           = false
+  override def getTotalFields: Int          = 1
+  override def getTypeClass: Class[Nothing] = classOf[Nothing]
+  override def getArity: Int                = 1
+  override def toString: String             = "MarkerTypeInfo"
+  override def equals(obj: Any): Boolean    = obj.isInstanceOf[MarkerTypeInfo.type]
+  override def hashCode(): Int              = 0
+
+}

modules/flink-common-api/src/test/scala-2/org/apache/flinkx/api/RecursiveTest.scala

@@ -0,0 +1,24 @@
+package org.apache.flinkx.api
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.util.FlinkRuntimeException
+import org.apache.flinkx.api.RecursiveTest.Node
+import org.apache.flinkx.api.serializers._
+import org.scalatest.Inspectors
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class RecursiveTest extends AnyFlatSpec with Matchers with Inspectors with TestUtils {
+
+  it should "fail fast with a comprehensive message when the derivation is recursive" in {
+    val exception = intercept[FlinkRuntimeException](implicitly[TypeInformation[Node]])
+    exception.getMessage shouldBe "Unsupported: recursivity detected in 'org.apache.flinkx.api.RecursiveTest.Node'."
+  }
+
+}
+
+object RecursiveTest {
+
+  case class Node(left: Option[Node], right: Option[Node])
+
+}

modules/flink-common-api/src/test/scala/org/apache/flinkx/api/SerializerTest.scala

@@ -86,11 +86,6 @@ class SerializerTest extends AnyFlatSpec with Matchers with Inspectors with Test
     roundtrip(ser, Nil)
   }
 
-  it should "derive recursively" in {
-    // recursive is broken
-    // val ti = implicitly[TypeInformation[Node]]
-  }
-
   it should "derive list" in {
     val ser = implicitly[TypeInformation[List[Simple]]].createSerializer(ec)
     all(ser, List(Simple(1, "a")))
@@ -286,8 +281,6 @@ object SerializerTest {
   case class P1(a: String) extends Param[String]
   case class P2(a: Int)    extends Param[Int]
 
-  case class Node(left: Option[Node], right: Option[Node])
-
   case class SimpleOption(a: Option[String])
 
   case class SimpleEither(a: Either[String, Int])