Alin Gheorghe created KAFKA-6350:
------------------------------------
Summary: File descriptors leak with persistent KeyValueStore
Key: KAFKA-6350
URL: https://issues.apache.org/jira/browse/KAFKA-6350
Project: Kafka
Issue Type: Bug
Components: streams
Affects Versions: 1.0.0, 0.10.2.1
Reporter: Alin Gheorghe
When using the low level processor API with persistent KV stores we observed
continuous increase in the number of SSTs on disk. The file descriptors remain
open until reaching the configured OS limit (100k in our case), when Kafka
Streams crashes with "Too many open files" exception. In our case this happens
regularly in about 17 hours of uptime. The commit interval is set to 5 seconds
and we never call it from our code.
Our topology consists in 1 source topic, 7 processors, 2 KV stores and 2 sink
topics. Retention policy is set to 2 days and the topics have 25 partitions.
Using the punctuation mechanism in Kafka Streams 1.0.0 we perform a cleanup
every 30 seconds which checks for keys that have not been updated for at least
20 minutes. The KV stores hold temporary user sessions which last for 5 minutes
and have about 50 updates (user actions).
2017-12-11 10:57:03
{code:none}
~ # lsof 1 | grep rocksdb.*.sst | wc -l
54
{code}
2017-12-11 11:45:31
{code:none}
~ # lsof 1 | grep rocksdb.*.sst | wc -l
6742
{code}
We use the following state store APIs: *all*, *get*, *delete*, *put*.
When switching to in memory state stores this obviously doesn't happen.
We have also tried to override the RocksDB parameter to *max_open_files* which
defaults to -1, but the configured values seems to be ignored and RocksDB
surpasses that threshold.
Sometimes the application crashes with different error which may or may not be
related. We will file a different Jira issue if it seems unrelated:
{code:none}
RocksDBExceptionJni::ThrowNew/StatusJni - Error: unexpected exception!
2017-12-12 11:37:25,758
[processing-08dd0a1b-a423-4a41-b7e0-995c9d985444-StreamThread-1] WARN
org.apache.kafka.streams.KafkaStreams - stream-client
[processing-08dd0a1b-a423-4a41-b7e0-995c9d985444]All stream threads have died.
The instance will be in error state and should be closed.
2017-12-12 11:37:25,758
[processing-08dd0a1b-a423-4a41-b7e0-995c9d985444-StreamThread-1] ERROR
com.X.Y.Z.ApiStreaming$ - [ApiStreaming] Thread 12 died with exception task
[0_257] Failed to flush state store eventQueueStore. Shutting down the entire
Kafka Streams process
org.apache.kafka.streams.errors.ProcessorStateException: task [0_257] Failed to
flush state store eventQueueStore
at
org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:248)
at
org.apache.kafka.streams.processor.internals.AbstractTask.flushState(AbstractTask.java:196)
at
org.apache.kafka.streams.processor.internals.StreamTask.flushState(StreamTask.java:324)
at
org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:304)
at
org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
at
org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:299)
at
org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:289)
at
org.apache.kafka.streams.processor.internals.AssignedTasks$2.apply(AssignedTasks.java:87)
at
org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:451)
at
org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:380)
at
org.apache.kafka.streams.processor.internals.TaskManager.commitAll(TaskManager.java:309)
at
org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1018)
at
org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:835)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744)
Caused by: java.lang.IllegalArgumentException: Illegal value provided for
SubCode.
at org.rocksdb.Status$SubCode.getSubCode(Status.java:109)
at org.rocksdb.Status.<init>(Status.java:30)
at org.rocksdb.RocksDB.flush(Native Method)
at org.rocksdb.RocksDB.flush(RocksDB.java:1743)
at
org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:435)
at
org.apache.kafka.streams.state.internals.RocksDBStore.flush(RocksDBStore.java:428)
at
org.apache.kafka.streams.state.internals.WrappedStateStore$AbstractStateStore.flush(WrappedStateStore.java:84)
at
org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.flush(InnerMeteredKeyValueStore.java:268)
at
org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.flush(MeteredKeyValueBytesStore.java:153)
at
org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:245)
... 14 more
{code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
