https://flink.apache.org/ logo
Join Slack
Powered by
Hi :slightly_smiling_face: I’m running Flink 1.17....
# troubleshooting
r
Hi 🙂 I’m running Flink 1.17. I’m trying to use both reactive mode and unaligned checkpoints. When flink restart itself it fail to start with job manager exception:
Copy code
Cannot rescale the given pointwise partitioner.\nDid you change the partitioner to forward or rescale?
The filnk job currently contain only ‘forward’ arrows between operators. This job contain no broadcast and no explicit shuffles. How this issue can be solved? or more broadly: how can flink be both autoscaled and contain stable checkpoints?
d
are you using broadcast in your pipeline?
r
no
When I use unaligned checkpoints alone, it works great - even when restarting job manager scaling etc When I use the same flink job and combine same unaligned checkpoint with reactive mode - when adding TMs flink fail to start with the issue above
d
it seems that you have chaining disabled, I don’t think that’s supported when you want to rescale from unaligned checkpoint
is there any special reason why you’ve chaining disabled?
r
Well it is not disabled, rather we start a lot on new chains using
Copy code
.startNewChain();
I remember that performance wise - splitting to multiple chains gave us better throughput. Ill try to put all operators in a single chain and will update
removed all the
Copy code
.startNewChain();
now I have only 2 operators still getting the same issue
d
This seems weird, there is no reason why this should break the chain
r
Ok, Ill try tomorrow to chain those 2 operators into 1. Thanks for the help so far!
Ok. I’m still not able to merge those 2 operators into one. I removed the rebalance and hpa. The problem consist - when scaling up/down flink fail on same error as above
It looks like the kafka-sink create a new operator 2nd operator created from line 247
d
I think I have a rough idea, will take a look later
r
thanks
d
I need more context on what’s before the kafka producer. I’ve tried that locally and everything gets properly chained. The example you’ve sent -> map followed by sink doesn’t do that for me.
👀 1
You can also share it via DM if you don’t want to do it publicly
r
thanks, sent in DM
Right now I still have 2 operators like before, using unaligned checkpoints and rebalance. both scaling up and down do work, while loading both from unaligned checkpoints and save points. Will keep testing and update
UC do work, but because removed the startNewChain(), the processing rate is hurt dramatically
As a result, I’ll try to use the changelog solution instead of UC to make service checkpoints more stable
d
• disabling chaining is sort of a scale out in this case, because each task will get a separate thread; it should be better to start TMs with more slots and increase parallelism all-together and let chaining intact • UC should work even with chaining disabled; if you’re able to reproduce a the error with it, it’s a bug that should be fixed (we’ll need a reproducer, because everything worked fine in my experiments) • changelog will hurt performance to certain extent (at the end it’s an extra work that needs to be done) • buffer debloating might do the trick for you as well, if there no heavy per-element code paths
r
Thanks! • Already tried 
buffer debloating
but it did’t change the buffer size and the checkpoint time stayed the same. we have applications with parallelism 1000+ and according to my understanding from docs, it is not suitable for such scale. images below • I do understand the downside of change log, but if it will stabilise the checkpoints it is worth it • A pre request to change log is setting savepoint-restore-mode to CLAIM - service fail applying this config. opened new thread regarding it
Eventually I did use unaligned checkpoints. I added SHUFFLE between the 2 operators - and then scaling works. It also turn out I missread out metrics - the processing rate were not decreased
2 Views
Open in Slack
PreviousNext
Powered by