https://flink.apache.org/ logo
Join Slack
Powered by
hey team, wanted to seek some advice on resuming f...
# troubleshooting
j
hey team, wanted to seek some advice on resuming from savepoints, schema evolution, and what's the best practice to avoid 
local class incompatible
 errors. what we're using:
Copy code
flink_version: 1.16
scala
i have a flink job that's parsing some raw json into a case class using 
org.json4s.native.JsonMethods._
. it's implemented extending a 
RichFlatMapFunction
:
Copy code
package com.myapp.input.transforms

import com.typesafe.scalalogging.LazyLogging
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.json4s._
import org.json4s.native.JsonMethods._

class ExtractFields extends RichFlatMapFunction[ProtoGeneratedScalaClass, MyJavaPojoClass] with LazyLogging {
  override def flatMap(inputRow: ProtoGeneratedScalaClass, out: Collector[MyJavaPojoClass]): Unit = {
    implicit val formats: Formats = DefaultFormats
    val request_payload = inputRow.data
    val extracted = parse(request_payload).extract[ScalaCaseClass] //parse using json4s into a ScalaCaseClass
    extracted.map(final_row => out.collect(final_row))
    }

  def transform(request: ScalaCaseClass): Array[MyJavaPojoClass] = {
      //extract fields from ScalaCaseClass and flatten it and put it into MyJavaPojoClass using setters
    }
}
right now, I want to add some columns to my final output 
MyJavaPojoClass
, so i'm modifying the java class to add additional fields and setters. I'm also modifying my intermediate case class that json4s parses into, 
ScalaCaseClass
, and adding additional setters in the 
transform
method. when i try to create a savepoint, update my application, and resume from it, i get the error:
Copy code
Caused by: java.io.InvalidClassException: com.myapp.input.transforms.ExtractFields; local class incompatible: stream classdesc serialVersionUID = -4599689154206395147, local class serialVersionUID = -1986061269051638537
	at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:699)
	at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2005)
	at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1852)
	at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2186)
	at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669)
	at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2431)
what's the idiomatic way to safely evolve the job to add more columns to support resuming from savepoints? should i somehow try to add the annotation 
@SerialVersionUID
and add the Serializable trait to 
com.myapp.input.transforms.ExtractFields
class and hardcode the serialversionUID? somehow that does not feel like it should be the right way to deal with this sort of situation. any advice would be greatly appreciated, or let me know if i should ask in the mailing list instead πŸ™
m
j
oh i did not πŸ€”
let me try that and see if that helps. do you know if i should be assigning a 
SerialVersionUID
to every case class or class in my job?
https://stackoverflow.com/questions/71445825/distributed-execution-with-apache-flink-and-deserialization-of-class-fields-me found this stack overflow question and the answerer did say it was a good idea to assign it
m
Every operator should have 
SerialVersionUID
- If you don't, you can run into this:
You can automatically restore from the savepoint as long as these IDs do not change. The generated IDs depend on the structure of your program and are sensitive to program changes. Therefore, it is highly recommended assigning these IDs manually.
j
ah i see, thanks so much for pointing me in this direction. missed this part of the documentation
will try adding them explicitly to see if i can restore from savepoints afterwards
m
Most likely that won't be possible unfortunately
Open in Slack
PreviousNext
Powered by