[NU-1848] Work around for broken compatibility with Flink < 1.19: env…

…ironment variable allowing to disable Flink TypeInfos
TouK · Nov 6, 2024 · ea72f05 · ea72f05
1 parent 062fdaf
commit ea72f05
Showing 1 changed file with 8 additions and 3 deletions.
diff --git a/...n/scala/pl/touk/nussknacker/engine/flink/api/typeinformation/FlinkTypeInfoRegistrar.scala b/...n/scala/pl/touk/nussknacker/engine/flink/api/typeinformation/FlinkTypeInfoRegistrar.scala
@@ -9,6 +9,8 @@ import java.util
 
 object FlinkTypeInfoRegistrar {
 
+  private val DisableFlinkTypesRegistrationEnvVarName = "NU_DISABLE_FLINK_TYPE_INFOS_REGISTRATION"
+
   private case class RegistrationEntry[T, K <: TypeInfoFactory[T]](klass: Class[T], factoryClass: Class[K])
 
   private val typesToRegister = List(
@@ -17,10 +19,13 @@ object FlinkTypeInfoRegistrar {
     RegistrationEntry(classOf[LocalDateTime], classOf[LocalDateTimeTypeInfoFactory]),
   )
 
-  def ensureBaseTypesAreRegistered(): Unit =
-    typesToRegister.foreach { base =>
-      register(base)
+  def ensureBaseTypesAreRegistered(): Unit = {
+    if (!Option(System.getenv(DisableFlinkTypesRegistrationEnvVarName)).exists(java.lang.Boolean.parseBoolean)) {
+      typesToRegister.foreach { base =>
+        register(base)
+      }
     }
+  }
 
   private def register(entry: RegistrationEntry[_, _ <: TypeInfoFactory[_]]): Unit = {
     val opt = Option(TypeExtractor.getTypeInfoFactory(entry.klass))