Hi Guozhang, I am using 10.2.1.
-Sameer. On Sat, Jul 29, 2017 at 12:05 AM, Guozhang Wang <wangg...@gmail.com> wrote: > 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 >