Sameer, This bug should be already fixed in trunk.
Which version of Kafka Streams are you running with? We can consider backport it and have a bug-fix release if it turns out to be a common issue. Guozhang On Fri, Jul 28, 2017 at 4:57 AM, Damian Guy <damian....@gmail.com> wrote: > 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.ChangeLoggingSegmentedBytesSto > re.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 > > > -- -- Guozhang
