Can anyone help me understand how flink handles the serialization and deserialization of Scala Case Classes? In the documentation, it appears that case classes are supported as special POJO types https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/datastream/fault-tolerance/serialization/types_serialization/ I tried setting up a simple unit test that utilizes a case class, with kryo serialization disabled, but it appears to be running into serialization errors

case class BasicCaseClass (id: Int)

class EnhancePackageMapperTest extends AnyFlatSpec with Matchers with BeforeAndAfter{
  val flinkCluster = new MiniClusterWithClientResource(
    new MiniClusterResourceConfiguration.Builder()
      .setNumberSlotsPerTaskManager(2)
      .setNumberTaskManagers(1)
      .build
  )

  before {
    flinkCluster.before()
  }

  after {
    flinkCluster.after()
  }

  "streaming data" should "collect values and sink correctly" in {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.getConfig.disableGenericTypes()

    val basicOne = BasicCaseClass(1)
    val basicTwo = BasicCaseClass(2)

    val testStream = env.fromElements(basicOne, basicTwo)
    val testStreamAction = testStream.map(x => x.id + 1)

    val result = env.execute()
  }
}

error code:

java.lang.UnsupportedOperationException: Generic types have been disabled in the ExecutionConfig and type com.ual.jarvis.BasicCaseClass is treated as a generic type.

I'm having some performance issues in my production application, and upon looking at the flame graph we're spending a good chunk of time with kryo serialization, i'm hoping to improve performance by navigating away from the kryo serializer.

Hi @Chris Payne, if you do not import TypeInformation instances for case classes via Scala implicits or pass them manually as explicit parameters to the Flink Scala API, Flink will fallback to Kryo as universal serializer. For example, this new library comes with special import: https://github.com/flink-extended/flink-scala-api#compile-times which generates TypeInformations for case class Foo (in this case, but can be anyy other Scala type of course)