Serializing Option fields in a state results in unexpected behaviour during tests

[lukas-mi]

I've encountered an odd behavior when testing my Flink application that does stateful processing: the state produced by a stateful function contains Option fields that when are None cannot be asserted correctly by the Scala test suite.

Setup:

• KeyedProcessFunction with a state of a case class type:

case class MyState(optField: Option[String])

object MyState {
  given typeInfo: TypeInformation[MyState] = deriveTypeInformation
}

• state is created/updated by applying CDC events on it, which are also Scala case classes
• state is created using value descriptor new ValueStateDescriptor[MyState]("myState", MyState.typeInfo)
• test harness is used in tests:

val fun = new MyStatefulFun()
val testHarness = new KeyedOneInputStreamOperatorTestHarness[String, CDCEvent, MyState](
      new KeyedProcessOperator(fun),
      _.eventId,
      Types.STRING
    )

• the state from the output

val states = testHarness.extractOutputValues().asScala
states.head shouldBe None

• the previous assertions results in test failure:

None was not equal to None
ScalaTestFailureLocation: my.package.functions.MyStatefulFun at (MyStatefulFun.scala:10)
Expected :None
Actual   :None

• it works fine when the field is set to Some("test-value")

Using:

• Scala @ 3.5.2
• Java @ 11.0.26 (Temurin)
• Flink @ 1.19.1
• Flink Scala API @ 1.19.1_1.2.1

What could be the issue?

[novakov-alexey]

I think this could be familiar issue. CC @arnaud-daroussin

Could you try to disable Kryo doing this in your test?

env.getConfig.disableGenericTypes()

[novakov-alexey]

Also, what if you create and setup explicit serialiser like here? https://github.com/flink-extended/flink-scala-api/blob/master/modules/examples/src/test/scala/org/example/CustomTriggerTests.scala#L73

[lukas-mi]

env.getConfig.disableGenericTypes() didn't help, however, passing the serializer to the test harness setup did the job 👍🏼

[novakov-alexey]

@lukas-mi ok, great. Does Flink fail with some Kryo error when disableGenericTypes() is on?

[lukas-mi]

It did not make a difference, assertion still failed with: None was not equal to None. And there was no Kryo error

[novakov-alexey]

I see. Interesting, thanks for checking.

@arnaud-daroussin do you think that using disableGenericTypes() in this case would fail with some error related to Kryo fallback is disabled? As we discussed earlier, None was not equal to None may happen when old Scala API is in CLASSPATH and Flink can fallback to Kryo silently. It seems this is not the case here. What do you think?

[arnaud-daroussin]

Hi,
Sorry for late reply, I was on leave last week.

I think disableGenericTypes() should allow to fail-fast with an explicit UnsupportedOperationException: Generic types have been disabled...
When I test to disable generic types on the CustomTriggerTests you pointed out, it is failing as expected. For example, if I replace line 73 by:

testHarness.getEnvironment.getExecutionConfig.disableGenericTypes()
testHarness.setup()

And when I test:

testHarness.getEnvironment.getExecutionConfig.disableGenericTypes()
testHarness.setup(serializer)

It is working, meaning no Kryo fallback occurred.

@lukas-mi, can you share with us your full test? To be sure we are not missing something.

Thanks