feat: fail-fast on Scala type resolution with Class

Author: arnaud-daroussin

README.md

@@ -187,6 +187,8 @@ However, this project cannot enforce TypeInformation usage in the Flink Java API
 
 Usage of this code may lead to silently fallback to Kryo.
 
+From Flink 1.19, a check is done to detect this misusage. To disable it, see [Disable fail-fast on Scala type resolution with Class feature flag](#disable-fail-fast-on-scala-type-resolution-with-class).
+
 > [!WARNING]  
 > Official `flink-scala` deprecated dependency contains Scala-specialized Kryo serializers. If this dependency is removed from the classpath (see [Supported Flink versions](#supported-flink-versions)), usage of Kryo with Scala classes leads to erroneous re-instantiations of `object` and `case object` singletons.
 > 
@@ -402,7 +404,7 @@ env
 
 In the [1.1.5 release](https://github.com/flink-extended/flink-scala-api/releases/tag/v1.18.1_1.1.5) the Case Class serialization process also [stores case class arity](https://github.com/flink-extended/flink-scala-api/pull/98/files#diff-e896c210d6a754cb3afb462aea34cca08f090330f6f3c663a64dfb5584fc3727R106) number to
 a savepoint. This was introduced to support Case Class schema evolution and allow to add new
-class fields with default values. However, unfortunatelly this is the breaking change to the Flink job state restore process. Flink job will fail, if
+class fields with default values. However, unfortunately this is the breaking change to the Flink job state restore process. Flink job will fail, if
 a savepoint used for the job restore was created by 1.1.4 or earlier releases.
 
 In order migrate to the 1.1.5 release version, one can use specially added environment variable:
@@ -412,12 +414,18 @@ To disable new savepoint format and be able to restore a Flink job with a savepo
 
 Example: `DISABLE_CASE_CLASS_ARITY_USAGE = true`
 
-To enbale new serialization logic set this variable to `false` or simply do not define this envrionment vairable.
+To enable new serialization logic set this variable to `false` or simply do not define this environment variable.
 
 Example: `DISABLE_CASE_CLASS_ARITY_USAGE = false`
 
 P.S. this flag can be deprecated in future when most of the users migrate to the latest library version.
 
+### Disable fail-fast on Scala type resolution with Class
+
+From [1.3.0 release](https://github.com/flink-extended/flink-scala-api/releases/tag/v1.19.1_1.3.0), a check is done to prevent misusage of Scala type resolution with `Class` which may lead to silently fallback to generic Kryo serializers.
+
+You can disable this check with the `DISABLE_FAIL_FAST_ON_SCALA_TYPE_RESOLUTION_WITH_CLASS` environment variable set to `true`.
+
 ## Release
 
 Create SBT file at ~/.sbt/1.0/sonatype.sbt with the following content:

modules/scala-api/src/main/scala/org/apache/flinkx/api/StreamExecutionEnvironment.scala

@@ -32,7 +32,7 @@ import org.apache.flink.api.connector.source.{Source, SourceSplit}
 import org.apache.flink.api.java.tuple
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable
 import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
-import org.apache.flink.configuration.{Configuration, ReadableConfig}
+import org.apache.flink.configuration.{ConfigOptions, Configuration, ReadableConfig}
 import org.apache.flink.core.execution.{JobClient, JobListener}
 import org.apache.flink.core.fs.Path
 import org.apache.flink.runtime.state.StateBackend
@@ -48,6 +48,7 @@ import java.net.URI
 import java.util
 import scala.jdk.CollectionConverters._
 import scala.language.implicitConversions
+import scala.util.Try
 
 @Public
 class StreamExecutionEnvironment(javaEnv: JavaEnv) {
@@ -826,7 +827,9 @@ object StreamExecutionEnvironment {
     * cluster.
     */
   def getExecutionEnvironment: StreamExecutionEnvironment = {
-    new StreamExecutionEnvironment(JavaEnv.getExecutionEnvironment)
+    val configuration = new Configuration
+    configureFailFastOnScalaTypeResolutionWithClass(configuration)
+    new StreamExecutionEnvironment(JavaEnv.getExecutionEnvironment(configuration))
   }
 
   /** Creates an execution environment that represents the context in which the program is currently executed.
@@ -835,6 +838,7 @@ object StreamExecutionEnvironment {
     *   Pass a custom configuration into the cluster.
     */
   def getExecutionEnvironment(configuration: Configuration): StreamExecutionEnvironment = {
+    configureFailFastOnScalaTypeResolutionWithClass(configuration)
     new StreamExecutionEnvironment(JavaEnv.getExecutionEnvironment(configuration))
   }
 
@@ -849,7 +853,9 @@ object StreamExecutionEnvironment {
     * [[setDefaultLocalParallelism(Int)]].
     */
   def createLocalEnvironment(parallelism: Int = JavaEnv.getDefaultLocalParallelism): StreamExecutionEnvironment = {
-    new StreamExecutionEnvironment(JavaEnv.createLocalEnvironment(parallelism))
+    val configuration = new Configuration
+    configureFailFastOnScalaTypeResolutionWithClass(configuration)
+    new StreamExecutionEnvironment(JavaEnv.createLocalEnvironment(parallelism, configuration))
   }
 
   /** Creates a local execution environment. The local execution environment will run the program in a multi-threaded
@@ -861,6 +867,7 @@ object StreamExecutionEnvironment {
     *   Pass a custom configuration into the cluster.
     */
   def createLocalEnvironment(parallelism: Int, configuration: Configuration): StreamExecutionEnvironment = {
+    configureFailFastOnScalaTypeResolutionWithClass(configuration)
     new StreamExecutionEnvironment(JavaEnv.createLocalEnvironment(parallelism, configuration))
   }
 
@@ -880,6 +887,7 @@ object StreamExecutionEnvironment {
   @PublicEvolving
   def createLocalEnvironmentWithWebUI(config: Configuration = null): StreamExecutionEnvironment = {
     val conf: Configuration = if (config == null) new Configuration() else config
+    configureFailFastOnScalaTypeResolutionWithClass(conf)
     new StreamExecutionEnvironment(JavaEnv.createLocalEnvironmentWithWebUI(conf))
   }
 
@@ -901,7 +909,9 @@ object StreamExecutionEnvironment {
     *   user-defined input formats, or any libraries, those must be provided in the JAR files.
     */
   def createRemoteEnvironment(host: String, port: Int, jarFiles: String*): StreamExecutionEnvironment = {
-    new StreamExecutionEnvironment(JavaEnv.createRemoteEnvironment(host, port, jarFiles: _*))
+    val configuration = new Configuration
+    configureFailFastOnScalaTypeResolutionWithClass(configuration)
+    new StreamExecutionEnvironment(JavaEnv.createRemoteEnvironment(host, port, configuration, jarFiles: _*))
   }
 
   /** Creates a remote execution environment. The remote environment sends (parts of) the program to a cluster for
@@ -924,7 +934,9 @@ object StreamExecutionEnvironment {
       parallelism: Int,
       jarFiles: String*
   ): StreamExecutionEnvironment = {
-    val javaEnv = JavaEnv.createRemoteEnvironment(host, port, jarFiles: _*)
+    val configuration = new Configuration
+    configureFailFastOnScalaTypeResolutionWithClass(configuration)
+    val javaEnv = JavaEnv.createRemoteEnvironment(host, port, configuration, jarFiles: _*)
     javaEnv.setParallelism(parallelism)
     new StreamExecutionEnvironment(javaEnv)
   }
@@ -949,7 +961,27 @@ object StreamExecutionEnvironment {
       config: Configuration,
       jarFiles: String*
   ): StreamExecutionEnvironment = {
+    configureFailFastOnScalaTypeResolutionWithClass(config)
     val javaEnv = JavaEnv.createRemoteEnvironment(host, port, config, jarFiles: _*)
     new StreamExecutionEnvironment(javaEnv)
   }
+
+  private def configureFailFastOnScalaTypeResolutionWithClass(configuration: Configuration): Unit = {
+    if (!isFailFastOnScalaTypeResolutionWithClassDisabled) {
+      val serializationOption = ConfigOptions.key("pipeline.serialization-config").stringType().asList().noDefaultValue()
+      val serializationConfig = configuration.getOptional(serializationOption).orElse(new util.ArrayList[String])
+      serializationConfig.add("scala.Product: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
+      serializationConfig.add("scala.Option: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
+      serializationConfig.add("scala.util.Either: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
+      serializationConfig.add("scala.Array: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
+      serializationConfig.add("scala.collection.Iterable: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
+      configuration.set(serializationOption, serializationConfig)
+    }
+  }
+
+  private lazy val isFailFastOnScalaTypeResolutionWithClassDisabled: Boolean =
+    sys.env
+      .get("DISABLE_FAIL_FAST_ON_SCALA_TYPE_RESOLUTION_WITH_CLASS")
+      .exists(v => Try(v.toBoolean).getOrElse(false))
+
 }

modules/scala-api/src/main/scala/org/apache/flinkx/api/typeinfo/FailFastTypeInfoFactory.scala

@@ -0,0 +1,30 @@
+package org.apache.flinkx.api.typeinfo
+
+import org.apache.flink.api.common.typeinfo.{TypeInfoFactory, TypeInformation}
+import org.apache.flink.util.FlinkRuntimeException
+import org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory.formatType
+
+import java.lang.reflect.Type
+import java.util
+import scala.jdk.CollectionConverters._
+
+class FailFastTypeInfoFactory extends TypeInfoFactory[Nothing] {
+
+  override def createTypeInfo(t: Type, params: util.Map[String, TypeInformation[_]]): TypeInformation[Nothing] =
+    throw new FlinkRuntimeException(
+      s"""You are using a 'Class' to resolve '${formatType(t, params)}' Scala type. flink-scala-api has no control over this kind of type resolution which may lead to silently fallback to generic Kryo serializers.
+         |Use type information instead: import 'org.apache.flinkx.api.serializers._' to make implicitly available in the scope required 'TypeInformation' to resolve Scala types.
+         |To disable this check, set 'DISABLE_FAIL_FAST_ON_SCALA_TYPE_RESOLUTION_WITH_CLASS' environment variable to 'true'.""".stripMargin
+    )
+
+}
+
+object FailFastTypeInfoFactory {
+
+  private def formatType(t: Type, params: util.Map[String, TypeInformation[_]]): String = if (params.isEmpty) {
+    t.getTypeName
+  } else {
+    params.keySet().asScala.mkString(s"${t.getTypeName}[", ", ", "]")
+  }
+
+}

modules/scala-api/src/test/scala/org/apache/flinkx/api/StreamExecutionEnvironmentTest.scala

@@ -5,9 +5,12 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.connector.source.Boundedness
 import org.apache.flink.api.connector.source.mocks.MockSource
 import org.apache.flink.api.java.typeutils.GenericTypeInfo
+import org.apache.flink.util.FlinkRuntimeException
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import scala.util.{Failure, Success, Try}
+
 class StreamExecutionEnvironmentTest extends AnyFlatSpec with Matchers with IntegrationTest {
 
   it should "create a stream from a source" in {
@@ -31,6 +34,24 @@ class StreamExecutionEnvironmentTest extends AnyFlatSpec with Matchers with Inte
     stream.dataType shouldBe typeInfo
   }
 
+  "From Flink 1.19, TypeInformation.of(Class)" should "fail-fast trying to resolve Scala type" in {
+    Try(Class.forName("org.apache.flink.configuration.PipelineOptions").getField("SERIALIZATION_CONFIG")) match {
+      case Failure(_) => // Before Flink 1.19: no fail-fast, exception happens at execution
+        implicit val typeInfo: TypeInformation[Option[Int]] = TypeInformation.of(classOf[Option[Int]])
+        val stream = env.fromElements(Some(1), None, Some(100))
+        val exception = intercept[UnsupportedOperationException] {
+          stream.executeAndCollect(3)
+        }
+        exception.getMessage should startWith("Generic types have been disabled in the ExecutionConfig and type scala.Option is treated as a generic type.")
+
+      case Success(_) => // From Flink 1.19: fail-fast at Scala type resolution
+        val exception = intercept[FlinkRuntimeException] {
+          TypeInformation.of(classOf[Option[Int]])
+        }
+        exception.getMessage should startWith("You are using a 'Class' to resolve 'scala.Option' Scala type.")
+    }
+  }
+
   // --------------------------------------------------------------------------
   //  mocks
   // --------------------------------------------------------------------------