Hey all 👋 — sorry for the vague question and the cross-post since I also posted this to the mailing list. I wanted to bring up the topic here to see if anyone in this channel may have some insights to share. I’m running Flink 2.2 .1 K8s Operator 1.15 with the ForSt async state backend, checkpoints in S3, working state around 6TB. With

upgradeMode

last-state

and using the adaptive scheduler we observe: • Downscale rescales (e.g. parallelism 24 -> 12) take roughly 2 hours from spec change to RUNNING (no events are processed in this time • Upscale rescales in the same flow are much faster, taking around 12 minutes The relevant ForSt / checkpointing settings are: •

use-ingest-db-restore-mode: true

•

incremental-restore-async-compact-after-rescale: true

•

state-recovery.claim-mode: CLAIM

•

NATIVE

savepoint format, incremental + compressed checkpoints •

pipeline.max-parallelism: 120

• ForSt cache

size-based-limit

350GB on a 500Gi gp3 volume per TM. • I’m running one slot per TM and have tried parallelisms between 12 and 24 My questions are: • Is ~2h for a downscale at this state size in line with what others see, or does it suggest a misconfiguration? Most of the time appears to be in restore. • Is the asymmetry where upscales are much faster also expected? • Given the asymmetry, are there settings or patterns that make autoscaling viable at ~1TB state beyond adjusting

job.autoscaler.scale-down.max-factor

and

scale-down.interval

? Or is the practical recommendation at this scale to keep parallelism static and rescale manually during low-traffic windows? Thank you!

encountered the same issue, we personally disabled autoscaling for now until it's resolved. We have 3TB of state

This isn’t directly related but what are your application startup times when redeploying with the same parallelisms? For me my stateful tasks take around 12 minutes to initialize with almost all of the time spent in the RocksDB

openDB()

method. I’ve verified this though the following log lines:

Starting opening base ForSt instance

Finished opening base ForSt instance

yes it's pretty much the same, it's around 5 minutes

5 minutes is not bad! I have another stateful job with ~3TB of state that is taking around the same time, 12 or so minutes. Do you mind sharing your parallelism? Are you running multiple slots pr JM?

s single TM has 4 slots, I have 16 TMs

Thanks! Yeah, our setup is a bit different — I’m running at a parallelism of 10 with 1 slot, so a total of 10 TMs

just out of curiosity, why 1 slot per task manager?

I’m wondering if we could decrease startup times while keeping an eye on resource usage by going with an approach similar to yours.

Maybe something else is your bottleneck, like your SSD IOPS or network limit. Do you use EBS? If so, what kind?

Yeah, we’re using EBS, it’s a gp3 volume using the xfs filesystem

yeah I thought so as well, but I noticed that during start up it downloads some data to local disk, which could be the bottleneck in some cases.

What settings are you using for EBS?

thanks, I may try this

hope it will help!

Yeah, sometimes — do you see this during checkpointing?

I don't think it happens only during checkpointing. Looks like that:

Caused by: java.io.InterruptedIOException: getFileStatus on <some s3 path>: com.amazonaws.SdkClientException: Unable to execute HTTP request: Timeout waiting for connection from pool

This I don’t see, are you using Hadoop or Presto?

hadoop

I’m using presto with a configured max-connections of 256

hmm interesting. I will try switching, thanks a lot!

Back on Flink 2.1 I had similar issues with Hadoop s3. After upgrading and switching to Presto it seems to have gotten better

well they did say it was experimental 🫠😂

Thanks again, I am seeing a very small decrease in initialization time when bumping the throughput and increasing the IOPS but it’s still not close to 5 minutes. I’m going to re-open the TM slot topic since I suspect your parallelism increase is really what is driving this speed. If it’s helpful, outside of parallelism we’ve been able to reduce initialization time by decreasing the amount of SST files. We did this by tuning compaction settings. https://nightlies.apache.org/flink/flink-docs-stable/docs/deployment/config/#forst-state-backend-configurable-options We increased the

target-file-size-base

,

max-size-level-base

and enabled

use-dynamic-size

. This is decreasing the amount of SST files while increasing the size. So far it’s gotten initialization time down from around 20 minutes at the default settings to the 12 or so we are seeing now. It’s a lazy process since it only processes changed data during compaction.

> I do see timeouts during checkpointing and it ends up being quite bad. Once I see this timeout all subsequent checkpoints will fail due to a metadata file Flink expects not being in S3. The job eventually restarts when it hits the failed checkpoint threshold We saw similar timeouts and "missing file" ...ended up being bad

/tmp

directory configuration setting. For us, the failed checkpoints was simply running out of /tmp disk space. You must specify the attached storage for all flink tmp stuffies.

-Djava.io.tmpdir=/mnt/ephemeral/flink/tmp

# Also set (for other stuff)
io.tmp=/mnt/ephemeral/flink/tmp

Hopefully that problem your facing is as simple as that!?