https://flink.apache.org/ logo
Join Slack
Powered by
Say that I have a Flink application with 3 statefu...
# troubleshooting
a
Say that I have a Flink application with 3 stateful operators, each one of those has a 
uid
explicitly set. I have operator chaining enabled, those 3 stateful operators got merged into one operator in the (looking in the Flink UI). For savepoint/checkpoint matters, are those operators still 3 individual "states" or are they merged into one and a new 
uidhash
is given to them? If so is this new 
uidhash
deterministic based on my other 3 explicitly set uids?
I have a large Flink application (40 operators) that get optimised to 19 operators. Whenever there's a simple code change that doesn't impact any of the operators the savepoint of the previous version can't be restored to the new version. The application is mostly DataStream, with all operators having `uid`defined and sinks we're mostly using the TableAPI.
we have messages like:
Copy code
2023-07-05 10:50:08,538 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 300de24cdbbd707fed887aa9ce07b752.
2023-07-05 10:50:08,538 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 318162186797e9c5e3c01e9b105a8138.
2023-07-05 10:50:08,538 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 8ea690fd176a34cabb4ba09b842582fe.
2023-07-05 10:50:08,538 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator a35239b8facc83c5c97033b34341a38f.
2023-07-05 10:50:08,538 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 90c480b8727bf01c9255cb030ab02e16.
2023-07-05 10:50:08,538 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 498315457fc0f4a921f4ce7f516b05a5.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 7dd3984cfd1c2ded45d7f0d71753b7d4.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 736613210be90d9e59f1aea2c70d0789.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator fe424978fd07d2d3c0cdd2a1d09825fe.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator f401222db8b6c49fefefd2e78e509f83.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator ce19a767600774cc16aa6516bf7703a4.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 79fab20e2b01b39d62af3df77b8a9062.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 9dbf8376c5e17d82e3730c4388a6dbea.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 9a52b21abcac86d7b26aef21fdb03d1e.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator abf7113fc554572651c91a4b6d80d37c.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 66e50b78965fa2bc7f0a7406643bb3ff.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 3a4e51bdd39e280dd766b201c45ed6ec.
2023-07-05 10:50:08,537 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator a52497a91aabfac700d35ea06ababb8d.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 94c0c07eb74a43995c3bdabdad21af6f.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 2ec9f0af02072ff1ac178e715c62da09.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator e3747dfba99def2ea3b0d0c3fd1fd2db.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator bace087d6d99e3d7ad2324a734344933.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 773f589724b770139a2dc399ca7a48b8.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator cfac49822173763e5cb27fac5a3ee423.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator d44b87617ac70bc1845d312b58e849ed.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 2e3c9d4fd9516e572a81a3e8def00ea3.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 4ff4caaeedb84d26c1ce1238562bef54.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 29e7e3766bdb937d96fbd6e7ac192979.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator b5fd71baa142253657a7025aa9d86b52.
2023-07-05 10:50:08,536 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator abdf967f03a7a10d21aa92230c241398.
2023-07-05 10:50:08,535 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 3bf3b89d13c1c9e758a5ac6d238a3c34.
2023-07-05 10:50:08,535 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator cbd9e1295569f4239cc8966373933fbf.
2023-07-05 10:50:08,535 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 66f71e30eb1bd9a1d4928aaae5ab37d2.
2023-07-05 10:50:08,535 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 275bfe14d0fc5223823645e6b51fc277.
2023-07-05 10:50:08,535 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 3c930a1da1b705212a8857c9651ea163.
2023-07-05 10:50:08,535 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 88d11dea6375f18ca09653704a3aa707.
2023-07-05 10:50:08,535 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator e1e5bdcf552bff9c5da90f7a39d3d88a.
2023-07-05 10:50:08,535 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 7c54c0cc3632045bcf3d4610cebb256b.
2023-07-05 10:50:08,535 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator 5464088241641d2f7c673a686dc4006d.
2023-07-05 10:50:08,534 [jobmanager-io-thread-2] INFO  org.apache.flink.runtime.checkpoint.Checkpoints              [] - Skipping savepoint state for operator fa851aadb1d85fb3f90f8f39c11a046d.
I would expect those operator hashes to be my own defined 
uid
but I'm not sure if everything just beocmes a uid-hash during a savepoint
Also: Are those IDs set by a per task base? Say I have 8 tasks and 5 stateful operators, would I get 40 of those 
Skipping savepoint state for operator
when a savepoint is not compatible and allowNonRestoredState is enabled?
d
When you have 3 stateful operators that are chained into one task, the individual operators remain distinct from the perspective of the states and their UIDs. Chaining is a networking optimization that doesn’t affect how the state is managed. Perhaps you have other stateful operators that lack explicit, user-provided UIDs — such as sources, sinks, windows, or joins? What does the whole topology look like?
a
plan
plan
Thanks for clarifying that 🙂 This is the current execution plan ☝️ It consists of 7 kafka-sources some stateful operators (for window joins and custom processing functions) and 5 sinks. The kafka sources use the datastream connector. For the sinks we're using the TableAPI to leverage the avro-confluent-schema-registry format.
We've disabled the auto-generation of UIDs and got some errors, but all related to operators without state. What I imagine can be a problem is using the Kafka Connector in the TableAPI as a sink, as it has state and from the best of my knowledge there's no way to assign a UID to it when using it. But I would imagine the sinks UID would be deterministic if there's no changes in the sink (no schema change an all)
@David Anderson Today I've enabled DEBUG logging in the job manager and manage to get the generate hash for every operator. With that in hand we reproduced the change that broke the savepoint, and indeed the problematic operator in the topology is the 
TableAPI Sink Committer
The thing is, that nothing changed in the pipeline just some minor methods renaming. All operators that we were able to set UID are restored correctly. Seems that relying on features from the TableAPI can cause problems when doing code upgrades to our Flink job. So is it fair to say that applications that require to always have precise control on state restore shouldn't use the TableAPI? The TableAPI is very convenient to use for sources and sinks, it would be a shame having to not use it because of the non deterministic operator hash, maybe I'm missing something in my application and that impacts the determinism, any clues?
d
In theory you can rely on the mechanisms described in FLIP-190 to upgrade table API jobs without losing any state. See https://docs.immerok.cloud/docs/how-to-guides/development/compiled-plan/ for an example of how to do this. However, for this to be effective, updates to the built-in table operators must respect the contract this FLIP relies on. I don't know if the sink committer was updated in a way that would have allowed you to upgrade without running into this problem -- in fact, I'm not sure that was possible. In general, the upgrade story for the Table API is still a work in progress.
a
Thanks for the answer @David Anderson, that's really helpful. From what I understood the plans are helpful when upgrading Flink versions as there can be new plan optimisations that could change the underlying operators of a SQL query. In my current scenario, we are not changing Flink versions nor anything related to the topology, we have the following setup: • DataStream Sources -> [DataStream transformations] -> table.fromDataStream -> table.insertInto • Some changes are deployed without state recovering issues. • Others are not, when there are state recovering problems I noticed that an Id in the Committer name changed ◦ from: 
'e2i__from_external_account_to_transfer_pix_edge[6]: Committer-187'"
◦ to: 
'e2i__from_external_account_to_transfer_pix_edge[4]: Committer-187'"
• There were no changes in the DataStream, not state changes, we didn't change te POJOs nor the Table's definition. If you could help me understand which part of our topology is non-deterministic (I guess it is the fromDatastream -> insertInto SQL part) and if we should migrate the final stages of our pipeline from SQL to DataStream for a production workload that ideally cannot start without previous state, I'd be very grateful! 🙂 Also, are you aware of any document that describes when can the Table/SQL API be used? The more I read the more caveats I find regarding state recovery, but also that there's a lot of work being done to make it better.
d
I can’t offer a theory that might explain what you’ve experienced. But I suspect that switching to using the DataStream API’s KafkaSink with ConfluentRegistryAvroSerializationSchema would help you avoid the problem.
a
@David Anderson thanks for the reply! We've done that and fixed the issue... but I was hoping to be able to use the TableAPI as it is very handy 😉
Open in Slack
PreviousNext
Powered by