Profiling First: The Tooling That Finds the Bottleneck
Before you can optimize anything, you need to see where time and memory are actually going. We lean on a handful of JVM profiling tools that cover both CPU and heap analysis. Async-profiler is our go-to for tracing CPU cycles and heap allocations; its flame graph output makes it easy to spot where Task Managers spend their time. For a quicker, interactive look at a running JVM, VisualVM provides a low-setup UI for live CPU and heap monitoring.
On the native memory side, jemalloc (the default Flink memory allocator since version 1.12) pairs with jeprof to let Task Managers and Job Managers dump memory profiles automatically. That combination has helped us watch memory trends over longer windows and catch a RocksDB leak in one of our applications. When we need deeper interpretation of a heap dump, Eclipse Memory Analyzer adds another layer of analysis—it was essential when we investigated out-of-memory failures in our GCS file sink.
These are the tools that worked for us, but the JVM ecosystem has plenty of others, from built-in commands like jmap to Java Flight Recorder.
Get Rid of Kryo Fallback Serialization
Flink supports several serializers out of the box. Scala case classes and Avro records are the common choices—not always the fastest, but good for developer experience. When neither fits a record type, Flink silently falls back to Kryo. That fallback is significantly slower than Flink's typical serializers, and you won’t notice it until you profile. When we ran async-profiler for unrelated performance work, the Kryo classes were taking up an alarming amount of space in the flame graph.
To force the issue, disable the fallback entirely with env.getConfig().disableGenericTypes();. Then fix whatever serialization failures surface. Two patterns we hit regularly:
- Scala's BigDecimal. Flink can serialize Java's
BigDecimalbut not Scala's. If you handle monetary values, default to the Java type. - Scala ADTs. Flink doesn't handle sealed traits with case objects. Scala enums are supported, so swap those in instead.
After we cleared out every Kryo dependency, throughput went up 20 percent.
Match Configuration to Your Load Profile
There's no one-size-fits-all Flink config. The right settings depend on the state and load of your pipeline. At Shopify, we see three distinct profiles: backfill (replaying history from the beginning of time until lag approaches zero), steady state (near-real-time consumption with minimal lag), and seasonal spikes like flash sales. The first two define how we tune most of our pipelines.
Backfill means massive backlog and a need for speed—you want throughput, not freshness, to catch up as fast as possible. That can mean processing tens of billions of messages in a couple of hours for large applications. Steady state flips the priority: minimize latency to keep outputs fresh. We end up running two different configuration profiles depending on which operational mode the pipeline is in.
With that context, a few parameters matter across both profiles:
- Input source partitions. Minimal lag at steady state means a small partition count might be plenty. But during a backfill, more partitions directly translate to more source throughput, so design your source counts with the replay scenario in mind.
- Back pressure. A healthy steady-state pipeline rarely sees back pressure. Backfills are different—bottlenecks show up clearly as red stages in the job graph UI. Use that window to identify and fix slow pipeline stages.
- Sink throttling. Your application code might not be the constraint; the sink itself can only handle so many writes. If you can't scale the sink (more database nodes, more Kafka partitions), consider lowering sink parallelism or reducing the number of outbound connections.
- Networking. Network buffers improve throughput by batching messages, but that comes with latency. Adding Task Managers and task slots can help parallelism, though a complex pipeline graph with several shuffle operations may need a bump to
taskmanager.memory.network.fraction. - Checkpointing. Stay aggressive with checkpoint frequency (
execution.checkpointing.interval) at steady state to keep recovery time low. During backfill, the overhead isn't worth it—reduce the frequency and check whether your Task Manager heap has enough headroom for checkpoint uploads if state is large. Incremental checkpoints (state.backend.incremental) help there too, and don't overlookexecution.checkpointing.timeoutif needed.
Find the Hot Spots in the Heap
Flink's File Sink writes to file systems and object stores such as HDFS, S3, or GCS, which Shopify uses. Configuration is straightforward, but making it both efficient and reliable takes some care.
The File Sink keeps an in-memory list of buckets, each determined by a BucketAssigner. A custom assigner can, for instance, derive a Hive-style partition (e.g., date=2021-01-01) from a timestamp field in each record.
The team initially added a File Sink to an existing DataStream naively:
val records: DataStream[Record] = …
val fileSink: SinkFunction[Record] = …
records.addSink(fileSink)
It passed tests. But when running a historical backfill in production, the application consumed all available Java heap and crashed — repeatedly, even after several memory increases. Some buffering per bucket was expected, but not tens of gigabytes.
Heap dumps taken just before the crashes and analyzed with Eclipse MAT pointed to two large objects dominating the heap:
The dominator tree highlighted the two HashMaps backing the File Sink buckets as the culprits:
Further inspection of the heap dump and logs revealed the root cause. With no data reshuffling, records destined for any bucket could land on any Task Manager. That meant each Task Manager maintained a large collection of buckets — regularly over 500 — and, because there wasn't enough data per Task Manager to trigger it, rolling and flushing files took much longer.
The fix was simple: key records by their partition string before sending them to the sink. This routes records for the same partition to the same Task Manager, so each one holds fewer buckets in memory and flushes files sooner. After the change, heap dump analysis showed a 90% decrease in active buckets per Task Manager.
This works well when partition cardinality is reasonably distributed. But a backfill with a few days of much heavier data can create skew. Adding hours to the partition key (e.g., date=2021-01-01-12) improves distribution and mitigates the problem. The broader lesson is that data locality affects all operators and sinks — logical shuffling for better parallelism can pay off in surprising places.
Put RocksDB on Fast Disks
RocksDB, the most popular Flink state backend, keeps some data in memory but persists most state to disk. When running large stateful applications, disk performance matters far more than it initially appears.
The team originally used Network File System (NFS) volumes for RocksDB state when deploying applications with little state, such as Kafka consumer offsets. No performance problems surfaced, and NFS added resiliency. But for an application with over 8 TB of state, they tried a GCP local SSD and saw processing rates improve roughly tenfold. The danger with a local SSD is its loss if the instance goes down, but Flink checkpoints and savepoints make state recovery straightforward.
Disable Dynamic Classloading If You Can
Flink loads user code in different ways: through the Java classpath (the JDK libraries plus Flink's /lib), via plugin components in /plugins, and through Dynamic User Code — classes in the JAR files of jobs submitted via REST, CLI, or the web UI. Dynamic User Code is loaded and unloaded per job, so lingering references to old classes can leak memory. Every job restart after a transient failure reloads it all.
The symptom was java.lang.OutOfMemoryError: Metaspace, and the heap screenshot showed Metaspace usage climbing with each restart. Putting the application on Java's common classpath — disabling dynamic classloading — stopped the growth entirely. That approach only suits Application Mode clusters that don't need to host multiple jobs, but where it applies, it removes a whole class of restart-related memory issues.
Check RocksDB’s Native Memory, Not Just the JVM
A particularly subtle memory issue emerged under conditions that were easy to miss: starting an application with a lot of state, waiting at least an hour, then manually killing a Task Manager container. A replacement should have joined the cluster via Kubernetes Deployment, and recovery should have followed. Instead, another Task Manager crashed with an out-of-memory error, starting an endless cycle of crashes and restarts.
Heap profiling with async-profiler and VisualVM showed nothing in the JVM, yet Kubernetes still killed pods for exceeding their memory limits. That pointed to native memory outside the JVM — specifically the RocksDB state backend. The team configured jemalloc to write periodic heap dumps to disk and analyzed them with jeprof. Just before another OOM error, the profile showed RocksDB allocating 6.74 GB against a Flink Managed Memory setting of 5.90 GB.
A RocksDB issue confirmed similar reports from many users over the years. The suggested workaround — disabling block cache through a custom RocksDBOptionsFactory — worked. Kubernetes stopped killing Task Managers after one was terminated.
Performance was unaffected. The only difference was in the time spent populating the cache; steady-state processing showed no gap between disabled and fully populated block caches. That also explained why the bug took an hour to reproduce: it took that long for the block cache to fill. Enabling RocksDB Native Metrics later confirmed it.
