Add feature flag to disable case class arity usage

src/main/scala/org/apache/flinkx/api/serializer/CaseClassSerializer.scala

@@ -77,8 +77,11 @@ abstract class CaseClassSerializer[T <: Product](
       createInstance(fields.toArray)
     }
 
+  private def isClassArityUsageDisabled =
+    sys.env.contains("DISABLE_CASE_CLASS_ARITY_USAGE")
+
   def serialize(value: T, target: DataOutputView): Unit = {
-    if (arity > 0)
+    if (arity > 0 && !isClassArityUsageDisabled)
       target.writeInt(value.productArity)
 
     (0 until arity).foreach { i =>
@@ -93,12 +96,15 @@ abstract class CaseClassSerializer[T <: Product](
   }
 
   def deserialize(reuse: T, source: DataInputView): T =
-    deserialize(source)
+    deserializeFromSource(source, isClassArityUsageDisabled)
+
+  def deserialize(source: DataInputView): T =
+    deserializeFromSource(source, isClassArityUsageDisabled)
 
-  def deserialize(source: DataInputView): T = {
+  private[api] def deserializeFromSource(source: DataInputView, classArityUsageDisabled: Boolean): T = {
     var i           = 0
     var fieldFound  = true
-    val sourceArity = if (arity > 0) Try(source.readInt()).getOrElse(arity) else 0
+    val sourceArity = if (arity > 0 && !classArityUsageDisabled) Try(source.readInt()).getOrElse(arity) else arity
     val fields      = new Array[AnyRef](arity)
     while (i < sourceArity && fieldFound) {
       Try(fieldSerializers(i).deserialize(source)) match {

src/test/scala/org/apache/flinkx/api/SchemaEvolutionTest.scala

@@ -1,35 +1,56 @@
 package org.apache.flinkx.api
 
-import org.apache.flinkx.api.SchemaEvolutionTest.{Click, Event, ClickEvent}
+import org.apache.flinkx.api.SchemaEvolutionTest.{Click, ClickEvent, Event, NoArityTest}
 import org.apache.flinkx.api.serializers._
+import org.apache.flinkx.api.serializer.ScalaCaseClassSerializer
 import org.apache.flink.core.memory.{DataInputViewStreamWrapper, DataOutputViewStreamWrapper}
-
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
 import java.io.ByteArrayOutputStream
 import java.nio.file.{Files, Path}
 
 class SchemaEvolutionTest extends AnyFlatSpec with Matchers {
-  private val ti = deriveTypeInformation[Event]
+  private val eventTypeInfo = deriveTypeInformation[Event]
+  private val arityTestInfo = deriveTypeInformation[NoArityTest]
   private val clicks =
     List(ClickEvent("a", "2021-01-01"), ClickEvent("b", "2021-01-01"), ClickEvent("c", "2021-01-01"))
 
   ignore should "generate blob for event=click+purchase" in {
     val buffer          = new ByteArrayOutputStream()
-    val eventSerializer = ti.createSerializer(null)
+    val eventSerializer = eventTypeInfo.createSerializer(null)
     eventSerializer.serialize(Click("p1", clicks), new DataOutputViewStreamWrapper(buffer))
     Files.write(Path.of("src/test/resources/click.dat"), buffer.toByteArray)
   }
 
   it should "decode click when we added view" in {
     val buffer = this.getClass.getResourceAsStream("/click.dat")
-    val click  = ti.createSerializer(null).deserialize(new DataInputViewStreamWrapper(buffer))
+    val click  = eventTypeInfo.createSerializer(null).deserialize(new DataInputViewStreamWrapper(buffer))
     click shouldBe Click("p1", clicks)
   }
+
+  ignore should "generate blob for no arity test" in {
+    val buffer          = new ByteArrayOutputStream()
+    val eventSerializer = arityTestInfo.createSerializer(null)
+    eventSerializer.serialize(NoArityTest(4, 3, List("test")), new DataOutputViewStreamWrapper(buffer))
+    Files.write(Path.of("src/test/resources/without-arity-test.dat"), buffer.toByteArray)
+  }
+
+  it should "decode class without arity info" in {
+    val buffer = this.getClass.getResourceAsStream("/without-arity-test.dat")
+    val serializer = arityTestInfo.createSerializer(null) match {
+      case s: ScalaCaseClassSerializer[_] => s
+      case s                              => fail(s"Serializer must be of CaseClassSerializer type, but was $s")
+    }
+    val decoded =
+      serializer.deserializeFromSource(new DataInputViewStreamWrapper(buffer), classArityUsageDisabled = true)
+    decoded shouldBe NoArityTest(4, 3, List("test"))
+  }
 }
 
 object SchemaEvolutionTest {
+  case class NoArityTest(field1: Long, field2: Long, field3: List[String] = Nil)
+
   sealed trait Event
   case class Click(
       id: String,