Remove failing CI test: a null serializer cannot happen (#269)

* Clean defensive code against a missing serializer: this cannot happen

* Remove failing CI test: a null serializer cannot happen

Author: arnaud-daroussin

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/CoproductSerializer.scala

@@ -8,8 +8,8 @@ import org.apache.flinkx.api.serializer.CoproductSerializer.CoproductSerializerS
 class CoproductSerializer[T](subtypeClasses: Array[Class[_]], subtypeSerializers: Array[TypeSerializer[_]])
     extends MutableSerializer[T] {
 
-  override val isImmutableType: Boolean = subtypeSerializers.forall(Option(_).exists(_.isImmutableType))
-  val isImmutableSerializer: Boolean    = subtypeSerializers.forall(Option(_).forall(s => s.duplicate().eq(s)))
+  override val isImmutableType: Boolean = subtypeSerializers.forall(_.isImmutableType)
+  val isImmutableSerializer: Boolean    = subtypeSerializers.forall(s => s.duplicate().eq(s))
 
   override def copy(from: T): T = {
     if (from == null || isImmutableType) {

modules/flink-2-api/src/main/scala/org/apache/flinkx/api/serializer/CoproductSerializer.scala

@@ -8,8 +8,8 @@ import org.apache.flinkx.api.serializer.CoproductSerializer.CoproductSerializerS
 class CoproductSerializer[T](subtypeClasses: Array[Class[_]], subtypeSerializers: Array[TypeSerializer[_]])
     extends MutableSerializer[T] {
 
-  override val isImmutableType: Boolean = subtypeSerializers.forall(Option(_).exists(_.isImmutableType))
-  val isImmutableSerializer: Boolean    = subtypeSerializers.forall(Option(_).forall(s => s.duplicate().eq(s)))
+  override val isImmutableType: Boolean = subtypeSerializers.forall(_.isImmutableType)
+  val isImmutableSerializer: Boolean    = subtypeSerializers.forall(s => s.duplicate().eq(s))
 
   override def copy(from: T): T = {
     if (from == null || isImmutableType) {

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

@@ -42,10 +42,8 @@ class CaseClassSerializer[T <: Product](
 
   private val nullPadding: Array[Byte] = if (super.getLength > 0) new Array(super.getLength) else EmptyByteArray
 
-  override val isImmutableType: Boolean = isCaseClassImmutable &&
-    fieldSerializers.forall(Option(_).exists(_.isImmutableType))
-  val isImmutableSerializer: Boolean =
-    fieldSerializers.forall(Option(_).forall(s => s.duplicate().eq(s)))
+  override val isImmutableType: Boolean = isCaseClassImmutable && fieldSerializers.forall(_.isImmutableType)
+  val isImmutableSerializer: Boolean = fieldSerializers.forall(s => s.duplicate().eq(s))
 
   // In Flink, serializers & serializer snapshotters have strict ser/de requirements.
   // Both need to be capable of creating one another.

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

@@ -21,11 +21,6 @@ class CaseClassSerializerTest extends AnyFlatSpec with Matchers {
   it should "be false when the content of one parameter is mutable" in {
     val mutableSerializer = new CaseClassSerializer[Mutable](classOf[Mutable], Array(StringSerializer.INSTANCE), false)
     val serializer = new CaseClassSerializer[OuterImmutable](classOf[OuterImmutable], Array(mutableSerializer), true)
-    mutableSerializer.isImmutableType should be(false)
-  }
-
-  it should "be false when missing information about one parameter" in {
-    val serializer = new CaseClassSerializer[Immutable](classOf[Immutable], Array(null), true)
     serializer.isImmutableType should be(false)
   }