Kafka Streams RocksDBException: Too Many Open Files Fix

Your Streams app runs fine for hours, then a thread dies mid-restore and the whole instance goes into ERROR state. The stack trace ends in RocksDB, not your topology. This is file-descriptor exhaustion, and it is almost never a leak in your code — it is the arithmetic of how many SST files a RocksDB state store keeps open multiplied by how many stores a single instance hosts.

This article is about the Too many open files variant specifically. It is distinct from InvalidStateStoreException (store migrated / not queryable) — that is a lifecycle problem; this is a resource-limit problem that kills the StreamThread outright.

1. The Error

The exact thing you pasted into Google, straight from the log:

org.apache.kafka.streams.errors.ProcessorStateException: Error opening store orders-store at location /tmp/kafka-streams/order-aggregator/0_12/rocksdb/orders-store
	at org.apache.kafka.streams.state.internals.RocksDBStore.openRocksDB(RocksDBStore.java:325)
	at org.apache.kafka.streams.state.internals.RocksDBStore.openDB(RocksDBStore.java:295)
	at org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:145)
	...
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:592)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:551)
Caused by: org.rocksdb.RocksDBException: While open a file for random read: /tmp/kafka-streams/order-aggregator/0_12/rocksdb/orders-store/000047.sst: Too many open files
	at org.rocksdb.RocksDB.open(Native Method)
	at org.rocksdb.RocksDB.open(RocksDB.java:239)
	at org.apache.kafka.streams.state.internals.RocksDBStore.openRocksDB(RocksDBStore.java:312)
	... 15 more

You will also see the OS-level cousins in the same window, because the FD limit is process-wide:

java.io.IOException: Too many open files
Caused by: org.rocksdb.RocksDBException: While lock file: /tmp/kafka-streams/.../LOCK: Too many open files

And the tell-tale broker/consumer side effect once the thread dies — the group rebalances because the instance dropped its partitions:

[Consumer clientId=order-aggregator-...-consumer, groupId=order-aggregator]
  Member ... sending LeaveGroup request to coordinator

Where it shows up: Kafka Streams 2.x, 3.x, and 4.x — the RocksDB embedding and its FD behavior have been constant across all of them. Typical setup: kafka-streams running on Linux (containers or bare metal) with a stateful topology — aggregations, windowed joins, KTable materializations. Managed Kafka doesn't matter; RocksDB runs in your JVM process, on your pod, against your OS limits. Default max_open_files in RocksDB is -1 (unlimited), so RocksDB will happily consume every descriptor the OS gives it.

2. How to Reproduce It

You do not need scale. You need a low ulimit and a topology with several partitions.

Minimal topology (kafka-streams 3.7.x):

// build.gradle: implementation 'org.apache.kafka:kafka-streams:3.7.1'
StreamsBuilder builder = new StreamsBuilder();

builder.stream("orders", Consumed.with(Serdes.String(), Serdes.String()))
    .groupByKey()
    .count(Materialized.as("orders-store"))   // one RocksDB store per partition (task)
    .toStream()
    .to("order-counts", Produced.with(Serdes.String(), Serdes.Long()));

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-aggregator");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Force the failure — create a wide topic and clamp the descriptor limit:

# 48 partitions => 48 tasks => 48 RocksDB stores on a single instance
kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic orders --partitions 48 --replication-factor 1

# Clamp the soft limit for the JVM you are about to launch
ulimit -n 1024
java -jar order-aggregator.jar

Each of the 48 tasks opens its own orders-store RocksDB instance. With the default RocksDB options each store holds open its CURRENT, MANIFEST, LOCK, OPTIONS, LOG, the current WAL, and one file descriptor per SST file in the LSM tree. A store with 15–20 SST files after some compaction burns ~25 descriptors. 48 stores × ~25 = ~1,200 FDs — over your 1,024 soft cap, plus the descriptors kafka-clients needs for broker sockets. The next RocksDB.open() throws.

Confirm it is FDs, not a leak — while it runs:

# PID of the streams JVM
watch -n1 'ls /proc/<pid>/fd | wc -l'
# Break it down: how many are .sst files?
ls -l /proc/<pid>/fd | grep -c '\.sst'

If that count climbs toward your ulimit -n and plateaus at the crash, it's descriptor exhaustion. If it climbs without bound across restores and rebalances even at steady partition count, you may also be hitting a genuine leak (see KAFKA-9603, fixed long ago but worth ruling out on old clients).

3. Why It Happens — Surface Level

RocksDB is a log-structured merge tree. Data lands in an in-memory memtable, gets flushed to immutable SST files, and background compaction merges those SST files into larger ones across levels. To serve a random read, RocksDB may need to open and hold a file handle for any SST file that could contain the key — and by default (max_open_files = -1) it caches every table handle it opens and never voluntarily closes them.

Now multiply. Kafka Streams creates one RocksDB store instance per state store per task, and there is one task per input partition (per sub-topology). A windowed store is worse: it's a segmented store, so a single logical store is several RocksDB databases, one per active time segment. Ten partitions of a windowed aggregation can be dozens of RocksDB instances, each with its own fleet of open SST files. The OS enforces a per-process descriptor ceiling (RLIMIT_NOFILE). When the sum of all those open handles plus your broker sockets crosses it, the next file open fails, and because it fails inside poll()/init(), it takes the StreamThread down with it.

4. Why It Happens — Under the Hood

Three mechanics stack up:

RocksDB table cache and max_open_files. RocksDB maintains a table cache keyed by file number. With max_open_files = -1 there is no eviction — every SST opened for a read stays open. Set a positive value and RocksDB treats it as an LRU bound: it will close the least-recently-used table handles to stay under the limit, at the cost of reopening them on a future read (a few extra syscalls). This single option is the difference between "descriptors grow with data volume" and "descriptors are capped per store."

Streams task fan-out. num.stream.threads controls parallelism, but it does not cap the number of stores — the number of tasks assigned to this instance does. A single instance that owns all partitions of a topic (common in a POC, or after every other instance leaves the group) hosts every store. During a rebalance with the eager assignor, an instance can transiently hold both its old and new assignments, briefly doubling open stores. Restoration makes it worse: to rebuild a store from its changelog, RocksDB writes and compacts aggressively, spawning more SST files exactly when many stores are opening at once.

The descriptor budget is shared. Those FDs compete with everything else the JVM holds open: one TCP socket per broker connection, consumer/producer internal files, your app's own file handles, JMX, and the JVM's own mmap'd regions. RocksDB with -1 is a greedy tenant that will expand until the whole process hits RLIMIT_NOFILE. The failing operation is often RocksDB simply because it's the largest consumer — the victim isn't always the cause.

The default container ulimit -n is frequently 1024. That number was fine for a stateless service. For a stateful Streams instance hosting dozens of RocksDB stores it's an accident waiting for enough partitions.

5. The Fix

Two independent levers. Apply both — they solve different halves of the problem.

Lever 1 — Bound RocksDB's open files with a config setter

Implement RocksDBConfigSetter and cap max_open_files (plus tame per-store memory so 50 stores don't each grab a fat block cache):

import org.apache.kafka.streams.state.RocksDBConfigSetter;
import org.rocksdb.*;

import java.util.Map;

public class BoundedRocksDBConfig implements RocksDBConfigSetter {

    // One shared block cache + write buffer manager across ALL stores in the process.
    // Static so every store instance shares the same 64 MB, not 64 MB EACH.
    private static final long TOTAL_OFF_HEAP_BYTES = 64L * 1024 * 1024;
    private static final Cache SHARED_CACHE = new LRUCache(TOTAL_OFF_HEAP_BYTES);
    private static final WriteBufferManager WRITE_BUFFER_MANAGER =
        new WriteBufferManager(TOTAL_OFF_HEAP_BYTES, SHARED_CACHE);

    @Override
    public void setConfig(final String storeName, final Options options,
                          final Map<String, Object> configs) {
        BlockBasedTableConfig table = (BlockBasedTableConfig) options.tableFormatConfig();

        table.setBlockCache(SHARED_CACHE);
        table.setCacheIndexAndFilterBlocks(true);            // count index/filter against the bound
        table.setPinTopLevelIndexAndFilter(true);
        options.setWriteBufferManager(WRITE_BUFFER_MANAGER);

        // The actual fix for "Too many open files": cap open table handles per store.
        options.setMaxOpenFiles(128);                        // was -1 (unbounded)

        // Fewer, larger SST files => fewer descriptors, less compaction churn.
        options.setMaxWriteBufferNumber(2);
        options.setWriteBufferSize(16L * 1024 * 1024);
    }

    @Override
    public void close(final String storeName, final Options options) {
        // Do NOT close SHARED_CACHE / WRITE_BUFFER_MANAGER here — they are shared and static.
    }
}

Wire it in:

props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG,
          BoundedRocksDBConfig.class);

Before / after — the one line that stops the bleeding:

- // default RocksDB options: max_open_files = -1 (unbounded)
+ options.setMaxOpenFiles(128);   // LRU-bound open SST handles per store

Pick the number deliberately: with max_open_files = 128 and ~50 stores you've capped RocksDB at ~6,400 descriptors regardless of data growth. Trade-off: too low and hot reads pay reopen syscalls; 100–256 is a sane band for most stores. Do not set it to a tiny value like 8 — you'll thrash.

Lever 2 — Raise the OS descriptor limit to match reality

Config setters cap per store; you still need headroom for the sum. Raise RLIMIT_NOFILE.

Docker Compose:

services:
  order-aggregator:
    image: order-aggregator:latest
    ulimits:
      nofile:
        soft: 100000
        hard: 100000

Kubernetes (init container, since pod specs can't set rlimits directly):

initContainers:
  - name: raise-nofile
    image: busybox
    securityContext: { privileged: true }
    command: ["sh", "-c", "ulimit -n 100000"]

systemd unit:

[Service]
LimitNOFILE=100000

Verify inside the running process — don't trust the host shell:

cat /proc/<pid>/limits | grep "open files"
# Max open files            100000    100000    files

Which lever when: In a POC or dev box, Lever 2 alone (bump ulimit -n) makes the error disappear and is fine. In production, do both: Lever 1 makes RocksDB usage bounded and predictable so a data-volume spike or a rebalance that piles stores onto one instance can't blow the limit, and Lever 2 gives generous headroom. Relying only on ulimit means you're one hot partition away from the same crash at a bigger number.

6. Best Practices & The Better Design

The failure class is "unbounded per-store resources × unbounded stores per instance." Bound both ends.

Bound RocksDB memory and files with shared, static resources. The config setter above uses a single static LRUCache and WriteBufferManager shared across all stores — this is the BoundedMemoryRocksDBConfig pattern from Confluent. Without it, each store gets its own 50 MB block cache by default; 50 stores = 2.5 GB of off-heap you didn't budget for, and each holds its own descriptors. Shared resources turn per-store cost into per-process cost you can actually reason about.

Cap stores per instance by spreading tasks. More instances (or num.standby.replicas planning) means fewer active tasks — and fewer stores — per JVM. Don't run 96 partitions of windowed state on a single fat instance; scale horizontally so no one process hosts the whole keyspace.

Prefer cooperative rebalancing. Kafka Streams uses the cooperative sticky assignor by default since 2.4. Confirm you haven't overridden it back to eager — eager revokes everything and can transiently double open stores during every rebalance, which is exactly when you're closest to the limit.

The "right way" config setter, complete:

public class ProductionRocksDBConfig implements RocksDBConfigSetter {
    private static final long OFF_HEAP = 128L * 1024 * 1024;
    private static final Cache CACHE = new LRUCache(OFF_HEAP);
    private static final WriteBufferManager WBM = new WriteBufferManager(OFF_HEAP, CACHE);

    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        BlockBasedTableConfig t = (BlockBasedTableConfig) options.tableFormatConfig();
        t.setBlockCache(CACHE);
        t.setCacheIndexAndFilterBlocks(true);
        t.setPinTopLevelIndexAndFilter(true);
        options.setWriteBufferManager(WBM);
        options.setMaxOpenFiles(256);
        options.setStatsDumpPeriodSec(0);          // quieter LOG files
        options.setInfoLogLevel(InfoLogLevel.ERROR_LEVEL);
    }

    @Override
    public void close(String storeName, Options options) { /* shared statics: leave open */ }
}

For stateless-heavy pipelines that only need dedup/lookups, consider whether you need RocksDB at all — in-memory stores (Stores.inMemoryKeyValueStore) hold zero SST files, at the cost of bounded, RAM-resident state and full changelog restore on restart.

7. How to Prevent It Long-Term

Monitor open descriptors as a first-class metric. Alert on process_open_fds / process_max_fds > 0.7. In containers, scrape /proc/<pid>/fd count or the JVM's UnixOperatingSystemMXBean.getOpenFileDescriptorCount(). This single alert catches the problem days before the crash.

Set the container ulimit -n as a platform convention. Bake LimitNOFILE / ulimits.nofile into your base image or Helm chart for every Streams service. Default 1024 should never reach production for a stateful app.

Standardize one RocksDBConfigSetter across services. Ship it as a shared library so max_open_files and shared-cache bounds are consistent, not rediscovered per team after an incident.

Load-test with production partition counts. The bug only appears at real fan-out. A CI/staging run with the actual partition count and a deliberately low ulimit -n (e.g. 4096) will surface descriptor pressure before customers do.

Watch state-store restore metrics. Restoration is the highest-FD-pressure moment. Track restore-latency and rebalance rate; frequent rebalances mean frequent store opens mean repeated FD spikes.

8. Key Takeaways

• RocksDBException: Too many open files is descriptor exhaustion, not a code bug: default max_open_files = -1 lets each RocksDB store hold one FD per SST file, forever.
• FD cost = (open files per store) × (stores per instance) × (segments, for windowed stores) + broker sockets. One instance owning all partitions is the danger case.
• Fix both ends: cap max_open_files (100–256) in a RocksDBConfigSetter, and raise the container ulimit -n (e.g. 100000). POC: ulimit alone. Production: both.
• Share one static LRUCache + WriteBufferManager across all stores, or 50 stores each grab their own 50 MB block cache and off-heap balloons.
• Alert on open_fds / max_fds; load-test at real partition counts with a low ulimit. This crash is 100% predictable and 100% preventable.