It is due to a bug. You should set StreamsConfig.STATE_DIR_CLEANUP_DELAY_MS_CONFIG to Long.MAX_VALUE - i.e., disabling it.
On Fri, 28 Jul 2017 at 10:38 Sameer Kumar <sam.kum.w...@gmail.com> wrote: > Hi, > > I am facing this error, no clue why this occurred. No other exception in > stacktrace was found. > > Only thing different I did was I ran kafka streams jar on machine2 a couple > of mins after i ran it on machine1. > > Please search for this string in the log below:- > org.apache.kafka.streams.processor.internals.StreamThread$1 for group > LICSp-4-25k failed on partition assignment > > > 2017-07-28 14:55:51 INFO StateDirectory:213 - Deleting obsolete state > directory 2_43 for task 2_43 > 2017-07-28 14:55:51 INFO StateDirectory:213 - Deleting obsolete state > directory 1_29 for task 1_29 > 2017-07-28 14:55:52 INFO StateDirectory:213 - Deleting obsolete state > directory 2_22 for task 2_22 > 2017-07-28 14:55:52 INFO StateDirectory:213 - Deleting obsolete state > directory 0_9 for task 0_9 > 2017-07-28 14:55:52 INFO StateDirectory:213 - Deleting obsolete state > directory 0_49 for task 0_49 > 2017-07-28 14:55:52 INFO StateDirectory:213 - Deleting obsolete state > directory 2_27 for task 2_27 > 2017-07-28 14:55:52 INFO StateDirectory:213 - Deleting obsolete state > directory 2_32 for task 2_32 > 2017-07-28 14:55:52 INFO StreamThread:767 - stream-thread [StreamThread-7] > Committing all tasks because the commit interval 5000ms has elapsed > 2017-07-28 14:55:52 INFO StreamThread:805 - stream-thread [StreamThread-7] > Committing task StreamTask 0_1 > 2017-07-28 14:55:52 ERROR StreamThread:813 - stream-thread [StreamThread-2] > Failed to commit StreamTask 1_35 state: > org.apache.kafka.streams.errors.ProcessorStateException: task [1_35] Failed > to flush state store lic3-deb-ci-25k > at > > org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:337) > at > > org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:72) > at > > org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188) > at > > org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:280) > at > > org.apache.kafka.streams.processor.internals.StreamThread.commitOne(StreamThread.java:807) > at > > org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:794) > at > > org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:769) > at > > org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:647) > at > > org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361) > Caused by: org.apache.kafka.streams.errors.ProcessorStateException: Error > while executing flush from store lic3-deb-ci-25k-201707280900 > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:354) > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flush(RocksDBStore.java:345) > at > org.apache.kafka.streams.state.internals.Segments.flush(Segments.java:138) > at > > org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.flush(RocksDBSegmentedBytesStore.java:117) > at > > org.apache.kafka.streams.state.internals.WrappedStateStore$AbstractWrappedStateStore.flush(WrappedStateStore.java:80) > at > > org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.flush(MeteredSegmentedBytesStore.java:111) > at > > org.apache.kafka.streams.state.internals.RocksDBWindowStore.flush(RocksDBWindowStore.java:92) > at > > org.apache.kafka.streams.state.internals.CachingWindowStore.flush(CachingWindowStore.java:120) > at > > org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:335) > ... 8 more > Caused by: org.rocksdb.RocksDBException: s > at org.rocksdb.RocksDB.flush(Native Method) > at org.rocksdb.RocksDB.flush(RocksDB.java:1642) > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:352) > ... 16 more > 2017-07-28 14:55:52 INFO StreamThread:767 - stream-thread > [StreamThread-12] Committing all tasks because the commit interval 5000ms > has elapsed > 2017-07-28 14:55:52 INFO StreamThread:390 - stream-thread [StreamThread-2] > Shutting down > 2017-07-28 14:55:52 INFO StreamThread:805 - stream-thread > [StreamThread-12] Committing task StreamTask 1_32 > 2017-07-28 14:55:52 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 0_0 > 2017-07-28 14:55:53 INFO StreamThread:767 - stream-thread > [StreamThread-15] Committing all tasks because the commit interval 5000ms > has elapsed > 2017-07-28 14:55:53 INFO StreamThread:805 - stream-thread > [StreamThread-15] Committing task StreamTask 0_32 > 2017-07-28 14:55:53 INFO StreamThread:767 - stream-thread [StreamThread-5] > Committing all tasks because the commit interval 5000ms has elapsed > 2017-07-28 14:55:53 INFO StreamThread:805 - stream-thread [StreamThread-5] > Committing task StreamTask 2_31 > 2017-07-28 14:55:53 INFO StreamThread:767 - stream-thread > [StreamThread-14] Committing all tasks because the commit interval 5000ms > has elapsed > 2017-07-28 14:55:53 INFO StreamThread:805 - stream-thread > [StreamThread-14] Committing task StreamTask 0_34 > 2017-07-28 14:55:53 INFO StreamThread:805 - stream-thread > [StreamThread-14] Committing task StreamTask 2_2 > 2017-07-28 14:55:53 INFO StreamThread:767 - stream-thread [StreamThread-8] > Committing all tasks because the commit interval 5000ms has elapsed > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 1_35 > 2017-07-28 14:55:53 INFO StreamThread:805 - stream-thread [StreamThread-8] > Committing task StreamTask 1_31 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 0_6 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 0_41 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 0_11 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 1_45 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 0_16 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 0_51 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 2_19 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 1_55 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 1_25 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-2] Closing task 1_30 > 2017-07-28 14:55:53 INFO StreamThread:554 - stream-thread [StreamThread-2] > Flushing state stores of task 0_0 > 2017-07-28 14:55:53 INFO StreamThread:554 - stream-thread [StreamThread-2] > Flushing state stores of task 1_35 > 2017-07-28 14:55:53 ERROR StreamThread:503 - stream-thread [StreamThread-2] > Failed while executing StreamTask 1_35 due to flush state: > org.apache.kafka.streams.errors.ProcessorStateException: task [1_35] Failed > to flush state store lic3-deb-ci-25k > at > > org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:337) > at > > org.apache.kafka.streams.processor.internals.AbstractTask.flushState(AbstractTask.java:199) > at > > org.apache.kafka.streams.processor.internals.StreamTask.flushState(StreamTask.java:421) > at > > org.apache.kafka.streams.processor.internals.StreamThread$4.apply(StreamThread.java:555) > at > > org.apache.kafka.streams.processor.internals.StreamThread.performOnTasks(StreamThread.java:501) > at > > org.apache.kafka.streams.processor.internals.StreamThread.flushAllState(StreamThread.java:551) > at > > org.apache.kafka.streams.processor.internals.StreamThread.shutdownTasksAndState(StreamThread.java:449) > at > > org.apache.kafka.streams.processor.internals.StreamThread.shutdown(StreamThread.java:391) > at > > org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:372) > Caused by: org.apache.kafka.streams.errors.ProcessorStateException: Error > while executing flush from store lic3-deb-ci-25k-201707280900 > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:354) > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flush(RocksDBStore.java:345) > at > org.apache.kafka.streams.state.internals.Segments.flush(Segments.java:138) > at > > org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.flush(RocksDBSegmentedBytesStore.java:117) > at > > org.apache.kafka.streams.state.internals.WrappedStateStore$AbstractWrappedStateStore.flush(WrappedStateStore.java:80) > at > > org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.flush(MeteredSegmentedBytesStore.java:111) > at > > org.apache.kafka.streams.state.internals.RocksDBWindowStore.flush(RocksDBWindowStore.java:92) > at > > org.apache.kafka.streams.state.internals.CachingWindowStore.flush(CachingWindowStore.java:120) > at > > org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:335) > ... 8 more > Caused by: org.rocksdb.RocksDBException: s > at org.rocksdb.RocksDB.flush(Native Method) > at org.rocksdb.RocksDB.flush(RocksDB.java:1642) > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:352) > ... 16 more > 2017-07-28 14:55:53 INFO StreamThread:554 - stream-thread [StreamThread-2] > Flushing state stores of task 0_6 > 2017-07-28 14:55:53 INFO StreamThread:554 - stream-thread [StreamThread-2] > Flushing state stores of task 0_41 > 2017-07-28 14:55:53 INFO StreamThread:767 - stream-thread [StreamThread-4] > Committing all tasks because the commit interval 5000ms has elapsed > 2017-07-28 14:55:53 INFO StreamThread:805 - stream-thread [StreamThread-4] > Committing task StreamTask 2_32 > 2017-07-28 14:55:53 ERROR ConsumerCoordinator:263 - User provided listener > org.apache.kafka.streams.processor.internals.StreamThread$1 for group > LICSp-4-25k failed on partition assignment > org.apache.kafka.streams.errors.ProcessorStateException: Error while > executing flush from store lic3-deb-ci-25k-201707280900 > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:354) > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flush(RocksDBStore.java:345) > at > org.apache.kafka.streams.state.internals.Segments.flush(Segments.java:138) > at > > org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.flush(RocksDBSegmentedBytesStore.java:117) > at > > org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.init(RocksDBSegmentedBytesStore.java:111) > at > > org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.init(ChangeLoggingSegmentedBytesStore.java:69) > at > > org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.init(MeteredSegmentedBytesStore.java:66) > at > > org.apache.kafka.streams.state.internals.RocksDBWindowStore.init(RocksDBWindowStore.java:77) > at > > org.apache.kafka.streams.state.internals.CachingWindowStore.init(CachingWindowStore.java:64) > at > > org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:86) > at > > org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:141) > at > > org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:864) > at > > org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:1237) > at > > org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.retryWithBackoff(StreamThread.java:1210) > at > > org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:967) > at > > org.apache.kafka.streams.processor.internals.StreamThread.access$600(StreamThread.java:69) > at > > org.apache.kafka.streams.processor.internals.StreamThread$1.onPartitionsAssigned(StreamThread.java:234) > at > > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:259) > at > > org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:352) > at > > org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:303) > at > > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:290) > at > > org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1029) > at > > org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:995) > at > > org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:592) > at > > org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361) > Caused by: org.rocksdb.RocksDBException: s > at org.rocksdb.RocksDB.flush(Native Method) > at org.rocksdb.RocksDB.flush(RocksDB.java:1642) > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:352) > ... 24 more > 2017-07-28 14:55:53 INFO StreamThread:390 - stream-thread [StreamThread-9] > Shutting down > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 2_33 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 1_4 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 2_38 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 1_9 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 2_43 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 1_49 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 1_19 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 1_54 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 2_23 > 2017-07-28 14:55:53 INFO StreamThread:1075 - stream-thread > [StreamThread-9] Closing task 1_59 > 2017-07-28 14:55:53 INFO StreamThread:554 - stream-thread [StreamThread-9] > Flushing state stores of task 2_33 > 2017-07-28 14:55:53 INFO StreamThread:554 - stream-thread [StreamThread-9] > Flushing state stores of task 1_4 > 2017-07-28 14:55:53 INFO StreamThread:554 - stream-thread [StreamThread-9] > Flushing state stores of task 2_38 > 2017-07-28 14:55:53 INFO StreamThread:554 - stream-thread [StreamThread-9] > Flushing state stores of task 1_9 > 2017-07-28 14:55:53 INFO StreamThread:554 - stream-thread [StreamThread-9] > Flushing state stores of task 2_43 > 2017-07-28 14:55:54 INFO StreamThread:767 - stream-thread [StreamThread-3] > Committing all tasks because the commit interval 5000ms has elapsed > 2017-07-28 14:55:54 INFO StreamThread:805 - stream-thread [StreamThread-3] > Committing task StreamTask 1_0 > 2017-07-28 14:55:54 INFO StreamThread:767 - stream-thread > [StreamThread-10] Committing all tasks because the commit interval 5000ms > has elapsed > 2017-07-28 14:55:54 INFO StreamThread:805 - stream-thread > [StreamThread-10] Committing task StreamTask 2_30 > 2017-07-28 14:55:54 INFO StreamThread:767 - stream-thread [StreamThread-6] > Committing all tasks because the commit interval 5000ms has elapsed > 2017-07-28 14:55:54 INFO StreamThread:805 - stream-thread [StreamThread-6] > Committing task StreamTask 1_34 > 2017-07-28 14:55:54 ERROR StreamThread:813 - stream-thread > [StreamThread-13] Failed to commit StreamTask 1_39 state: > org.apache.kafka.streams.errors.ProcessorStateException: task [1_39] Failed > to flush state store lic3-deb-ci-25k > at > > org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:337) > at > > org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:72) > at > > org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188) > at > > org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:280) > at > > org.apache.kafka.streams.processor.internals.StreamThread.commitOne(StreamThread.java:807) > at > > org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:794) > at > > org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:769) > at > > org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:647) > at > > org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361) > Caused by: org.apache.kafka.streams.errors.ProcessorStateException: Error > while executing flush from store lic3-deb-ci-25k-201707280900 > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:354) > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flush(RocksDBStore.java:345) > at > org.apache.kafka.streams.state.internals.Segments.flush(Segments.java:138) > at > > org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.flush(RocksDBSegmentedBytesStore.java:117) > at > > org.apache.kafka.streams.state.internals.WrappedStateStore$AbstractWrappedStateStore.flush(WrappedStateStore.java:80) > at > > org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.flush(MeteredSegmentedBytesStore.java:111) > at > > org.apache.kafka.streams.state.internals.RocksDBWindowStore.flush(RocksDBWindowStore.java:92) > at > > org.apache.kafka.streams.state.internals.CachingWindowStore.flush(CachingWindowStore.java:120) > at > > org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:335) > ... 8 more > Caused by: org.rocksdb.RocksDBException: s > at org.rocksdb.RocksDB.flush(Native Method) > at org.rocksdb.RocksDB.flush(RocksDB.java:1642) > at > > org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:352) > ... 16 more > 2017-07-28 14:55:54 INFO StreamThread:390 - stream-thread > [StreamThread-13] Shutting down > > > Regards, > -Sameer >
