Serialization issue

[Ajay Babu]

Hi Team,

I implemented a custom HttpEnricher on Nussknacker1.18.x, and created a streaming job and deployed in Flink 1.19.x cluster.

When one of the task manager in flink cluster down, flink trying to re-start the job with remaining task-managers, I am facing an issue as attached. Job getting canceled after retries. With  in-built nussknacker processers not noticed this issue.  I tried multiple ways but not getting fix it. I have attached flow and custom classes. Please help how to fix this.

Thanks,

Ajay Babu Maguluri.

[Ajay Babu]

Hi Team,

Any help here.

Thanks,

Ajay Babu Maguluri.

[Arkadiusz Burdach]

Hi,

Which scala version are you using? By default, Nussknacker 1.18.1 can only be run with scala 2.12. In stacktrace that you've attached, I see that Flink tries to deserialize pl.touk.nussknacker.engine.process.typeinformation.internal.typedobject.TypedScalaMapSerializer$$anon$1 class. It is a strange situation, because in the Nussknacker distribution, class TypedScalaMapSerializer has no anonymous inner classes. I guess that some automatic  Flink's mechanism sees that TypedScalaMapSerializer is a case class and tries to automatically deserialize it.  It is only my guess. To help you better, I would need a working example reproducing the issue. The best if it were a test case in the Nussknacker repo or repo with docker-compose and all necessary jars.

For the wider picture, TypedScalaMapSerializer is Nussknacker custom serializer for scala maps. We use it in TypingResultAwareTypeInformationDetection mechanism which prepares Flink serializers for data that is passed through the data stream.

Best regards,

Arek

[Ajay Babu]

Hi Ark,

Thanks for the helpful reply.

I have noticed the same issue with the built-in Nussnacker enrichers. I’ve attached a flow that includes a lookup-enricher. When I run this flow on a Flink cluster (version 1.19.1) with two task managers, and one task manager goes down, I get the following error:

Caused by: java.lang.RuntimeException: The class 'pl.touk.nussknacker.engine.process.typeinformation.internal.typedobject.TypedScalaMapSerializer$$anon$1' is not instantiable: The class is an inner class, but not statically accessible.

The issue is the same as when using a custom enricher (built with Scala 2.12). I have attached the scenario properties as well. When the I/O mode is Asynchronous, I face this issue; when it was Synchronous, there was no issue.

Please let me know if you are expecting any files to be shared. We can also have a call to review the issue in our test lab if that is fine with you.

Thanks & Regards

Ajay Babu Maguluri.

--
You received this message because you are subscribed to the Google Groups "Nussknacker" group.
To unsubscribe from this group and stop receiving emails from it, send an email to nussknacker...@googlegroups.com.
To view this discussion, visit https://groups.google.com/d/msgid/nussknacker/aaa45d0b-d34a-40e1-a12b-1b88ddabc67en%40googlegroups.com.