Hi Team! Every time I run my streaming job after 5-6 hours I get this error:

2023-08-02 12:13:04
java.lang.IllegalStateException: The library registration references a different set of library BLOBs than previous registrations for this job:
old:[p-275820cb9b5e36c9f3e1e5483e93d0b808fe257e-3978d0a9e10fc353bed49fdce5c5c6a9]
new:[p-275820cb9b5e36c9f3e1e5483e93d0b808fe257e-810951a32f2c50289fad972fb5701863]
	at org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager$ResolvedClassLoader.verifyClassLoader(BlobLibraryCacheManager.java:419)
	at org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager$ResolvedClassLoader.access$500(BlobLibraryCacheManager.java:359)
	at org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager$LibraryCacheEntry.getOrResolveClassLoader(BlobLibraryCacheManager.java:235)
	at org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager$LibraryCacheEntry.access$1100(BlobLibraryCacheManager.java:202)
	at org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager$DefaultClassLoaderLease.getOrResolveClassLoader(BlobLibraryCacheManager.java:336)
	at org.apache.flink.runtime.taskmanager.Task.createUserCodeClassloader(Task.java:1041)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:624)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
	at java.base/java.lang.Thread.run(Unknown Source)

I read that this could be due to corrupted checkpoints. How I can avoid this problem? I don’t have any flink classes that I overwrote, only custom java UDF. I use Flink 1.17.1. List of jars that are used:

flink-cep-1.17.1.jar              flink-json-1.17.1.jar                         flink-table-api-java-uber-1.17.1.jar   log4j-api-2.17.1.jar
flink-connector-files-1.17.1.jar  flink-s3-fs-hadoop-1.17.1.jar                 flink-table-planner-loader-1.17.1.jar  log4j-core-2.17.1.jar
flink-csv-1.17.1.jar              flink-scala_2.12-1.17.1.jar                   flink-table-runtime-1.17.1.jar         log4j-slf4j-impl-2.17.1.jar
flink-dist-1.17.1.jar             flink-sql-avro-confluent-registry-1.17.1.jar  franz-udfs-1.0-SNAPSHOT.jar
flink-gs-fs-hadoop-1.17.1.jar     flink-sql-connector-kafka-1.17.1.jar          log4j-1.2-api-2.17.1.jar

Thank you!

Does this exception start popping up randomly or after some restarts? I would suspect that the job restarted and something on the TaskManagers changed

Hi! @rmetzger Thank you for your response! This message starts popping after somre taskmanager restarts, but nothing have changed in configuration. In jobmanager I can find only this exception.

Is it possible that the contents of

franz-udfs-1.0-SNAPSHOT.jar

(or any other jar), or the underlying docker image, or any resources that have been fetched externally have changed in the meantime?

Thank you for response! No, we don’t change anything in featureee, when this happens. Code, image stay same (franz-udfs-1.0-SNAPSHOT.jar including). It usually happens at night when job was already running for 7-8 hours. Can you recommend how to find root cause?

I would dive into

BlobLibraryCacheManager

and related code. it seems that the

BlobLibraryCacheManager

is checking that all the jars stay the same, however there seems to be a change. Afaik, the serialized jobgraph created by the main() method of your Flink job is also shipped to the TaskManagers using the BlobLibraryCM ... and from the name of the blob

p-275820cb9b5e36c9f3e1e5483e93d0b808fe257e-810951a32f2c50289fad972fb5701863

this could be it (I would validate that). Then you need to find out why the main() method regenerates the job graph (add a logging statement and see if it gets re-executed?) The job graph gets re-generated if the HA config maps have been modified. The Flink K8s operator is doing that. Are you using it? Maybe that's causing it?

Thank you for answer, we will look into this problem. We are using k8s operator and config maps to transfer python job code into container. HA was not used for this job, but probably problem is with config map, will look into that.