You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Sachin Mittal <sj...@gmail.com> on 2017/03/25 08:20:20 UTC
kafka streams 0.10.2 failing while restoring offsets
Hi,
So recently we fixed the deadlock issue and also built the streams jar by
copying the rocks db configs from trunk.
So we don't get any deadlock issue now and also we see that the wait time
of CPU cores stays around 5% (down from 50% earlier).
However we now get a new exception which is not handled by streams
application and causes the instance to shutdown.
org.apache.kafka.streams.errors.StreamsException: stream-thread
[StreamThread-2] Failed to rebalance
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:622)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:378)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
Caused by: java.lang.IllegalStateException: task [0_9] Log end offset of
new-part-advice-key-table-changelog-9 should not change while restoring:
old end offset 647352, current offset 647632
at
org.apache.kafka.streams.processor.internals.ProcessorStateManager.restoreActiveState(ProcessorStateManager.java:240)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
at
org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:193)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
at
org.apache.kafka.streams.processor.internals.AbstractProcessorContext.register(AbstractProcessorContext.java:99)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
at
org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.init(RocksDBSegmentedBytesStore.java:101)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
at
org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.init(ChangeLoggingSegmentedBytesStore.java:68)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
at
org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.init(MeteredSegmentedBytesStore.java:66)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
at
org.apache.kafka.streams.state.internals.RocksDBWindowStore.init(RocksDBWindowStore.java:76)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
at
org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:86)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
at
org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:141)
~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
What I check from logs is this
DEBUG 2017-03-25 02:07:24,499 [StreamThread-2]:
org.apache.kafka.streams.processor.internals.StreamThread - stream-thread
[StreamThread-2] creating new task 0_9
So it creates the task at this time.
To create the local state store from the chnagelog topic it starts at
DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
org.apache.kafka.clients.consumer.KafkaConsumer - Subscribed to
partition(s): new-part-advice-key-table-changelog-9
DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to end of
partition new-part-advice-key-table-changelog-9
DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset for
partition new-part-advice-key-table-changelog-9 to latest offset.
DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
org.apache.kafka.clients.consumer.internals.Fetcher - Handling
ListOffsetResponse response for new-part-advice-key-table-changelog-9.
Fetched offset 647352, timestamp -1
DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to beginning of
partition new-part-advice-key-table-changelog-9
DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset for
partition new-part-advice-key-table-changelog-9 to earliest offset.
and process is over at
DEBUG 2017-03-25 02:10:21,225 [StreamThread-2]:
org.apache.kafka.clients.consumer.internals.Fetcher - Sending fetch for
partitions [new-part-advice-key-table-changelog-9] to broker
192.168.73.199:9092 (id: 5 rack: null)
DEBUG 2017-03-25 02:10:21,230 [StreamThread-2]:
org.apache.kafka.clients.consumer.KafkaConsumer - Unsubscribed all topics
or patterns and assigned partitions
And the exception is thrown at:
ERROR 2017-03-25 02:10:21,232 [StreamThread-2]:
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - User
provided listener
org.apache.kafka.streams.processor.internals.StreamThread$1 for group
new-part-advice failed on partition assignment
java.lang.IllegalStateException: task [0_9] Log end offset of
new-part-advice-key-table-changelog-9 should not change while restoring:
old end offset 647352, current offset 647632
So you can clearly see that while restoring the state some other thread (on
same or other instance) did commit some more offset for this change long
partition so in the end of the process two offsets did not match. I think
this is fairly a reasonable scenario and while restoring the state it
should also consider any added offsets and not assume that this is the only
thread processing that partition. It may have been some other instance did
commit some more offsets while this thread is trying to restore the state.
So I feel this exception should be handled and not thrown all the way to
the streams.
What do you all think.
Thanks
Sachin
Re: kafka streams 0.10.2 failing while restoring offsets
Posted by Sachin Mittal <sj...@gmail.com>.
Well lets say previous thread was processing a large block and while
processing it got bumped out.
So new thread which got this partition may have different offset when it
starts the restore and by the time it completes the restore.
If that is the case then it should just ignore that task for that cycle and
move on to next task.
When it returns again it can again try to restore the state and if by that
time there is no other thread processing the partition it can start
processing.
I see no reason to raise the exception and kill the thread entirely.
Thanks
Sachin
On Mon, Mar 27, 2017 at 3:56 PM, Damian Guy <da...@gmail.com> wrote:
> Yes, but we don't know why it is still processing the data. We don't want
> to have multiple processes acting on the same tasks, hence the exception.
> What if for some reason the other task is processing a large backlog, how
> long do we wait before we give up?
>
> I think in this case the exception is the right thing to do
>
> On Mon, 27 Mar 2017 at 09:24 Sachin Mittal <sj...@gmail.com> wrote:
>
> Hi,
> These are the logs
> https://www.dropbox.com/s/2t4ysfdqbtmcusq/complete_84_85_87_log.zip?dl=0
>
> I think this may not always be the case especially if previous owner is on
> a different machine.
>
> Say it is processing and it takes more than the poll timeout to process or
> commit the offset.
>
> The group bumps this thread and assigns its task to a different thread on
> maybe a different machine.
>
> All this while this client may be pushing the changelog data and other
> thread restoring the state from the same partition.
>
> So between the time it starts and it seeks to the end of the changelog it
> may be possible that previous thread which was still in process since it
> did not know it got bumped out added some more data to that changelog.
>
> Only when previous thread tries to commit the offset it gets to know that
> it is no longer the owner of the partition and then issues a rejoin
> request.
>
> I think in such a case should be handled within streams application.
>
> Thanks
> Sachin
>
>
>
>
>
> On Mon, Mar 27, 2017 at 1:25 PM, Damian Guy <da...@gmail.com> wrote:
>
> > Hi Sachin,
> >
> > This should not happen. The previous owner of the task should have
> stopped
> > processing before the restoration begins. So if this is happening, then
> > that signals a bug. Do you have more logs?
> >
> > Thanks,
> > Damian
> >
> > On Sat, 25 Mar 2017 at 08:20 Sachin Mittal <sj...@gmail.com> wrote:
> >
> > > Hi,
> > > So recently we fixed the deadlock issue and also built the streams jar
> by
> > > copying the rocks db configs from trunk.
> > > So we don't get any deadlock issue now and also we see that the wait
> time
> > > of CPU cores stays around 5% (down from 50% earlier).
> > >
> > > However we now get a new exception which is not handled by streams
> > > application and causes the instance to shutdown.
> > >
> > > org.apache.kafka.streams.errors.StreamsException: stream-thread
> > > [StreamThread-2] Failed to rebalance
> > > at
> > >
> > > org.apache.kafka.streams.processor.internals.StreamThread.runLoop(
> > StreamThread.java:622)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > at
> > >
> > > org.apache.kafka.streams.processor.internals.
> > StreamThread.run(StreamThread.java:378)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > Caused by: java.lang.IllegalStateException: task [0_9] Log end offset
> of
> > > new-part-advice-key-table-changelog-9 should not change while
> restoring:
> > > old end offset 647352, current offset 647632
> > > at
> > >
> > > org.apache.kafka.streams.processor.internals.ProcessorStateManager.
> > restoreActiveState(ProcessorStateManager.java:240)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > at
> > >
> > > org.apache.kafka.streams.processor.internals.ProcessorStateManager.
> > register(ProcessorStateManager.java:193)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > at
> > >
> > > org.apache.kafka.streams.processor.internals.AbstractProcessorContext.
> > register(AbstractProcessorContext.java:99)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > at
> > >
> > > org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.
> > init(RocksDBSegmentedBytesStore.java:101)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > at
> > >
> > > org.apache.kafka.streams.state.internals.
> ChangeLoggingSegmentedBytesSto
> > re.init(ChangeLoggingSegmentedBytesStore.java:68)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > at
> > >
> > > org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.
> > init(MeteredSegmentedBytesStore.java:66)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > at
> > >
> > > org.apache.kafka.streams.state.internals.RocksDBWindowStore.init(
> > RocksDBWindowStore.java:76)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > at
> > >
> > > org.apache.kafka.streams.processor.internals.AbstractTask.
> > initializeStateStores(AbstractTask.java:86)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > > at
> > >
> > > org.apache.kafka.streams.processor.internals.
> > StreamTask.<init>(StreamTask.java:141)
> > > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > >
> > > What I check from logs is this
> > > DEBUG 2017-03-25 02:07:24,499 [StreamThread-2]:
> > > org.apache.kafka.streams.processor.internals.StreamThread -
> > stream-thread
> > > [StreamThread-2] creating new task 0_9
> > > So it creates the task at this time.
> > >
> > > To create the local state store from the chnagelog topic it starts at
> > >
> > > DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> > > org.apache.kafka.clients.consumer.KafkaConsumer - Subscribed to
> > > partition(s): new-part-advice-key-table-changelog-9
> > > DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> > > org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to end of
> > > partition new-part-advice-key-table-changelog-9
> > > DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> > > org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset
> > for
> > > partition new-part-advice-key-table-changelog-9 to latest offset.
> > > DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> > > org.apache.kafka.clients.consumer.internals.Fetcher - Handling
> > > ListOffsetResponse response for new-part-advice-key-table-changelog-9.
> > > Fetched offset 647352, timestamp -1
> > > DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> > > org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to beginning
> > of
> > > partition new-part-advice-key-table-changelog-9
> > > DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> > > org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset
> > for
> > > partition new-part-advice-key-table-changelog-9 to earliest offset.
> > >
> > > and process is over at
> > > DEBUG 2017-03-25 02:10:21,225 [StreamThread-2]:
> > > org.apache.kafka.clients.consumer.internals.Fetcher - Sending fetch
> for
> > > partitions [new-part-advice-key-table-changelog-9] to broker
> > > 192.168.73.199:9092 (id: 5 rack: null)
> > > DEBUG 2017-03-25 02:10:21,230 [StreamThread-2]:
> > > org.apache.kafka.clients.consumer.KafkaConsumer - Unsubscribed all
> > topics
> > > or patterns and assigned partitions
> > >
> > > And the exception is thrown at:
> > > ERROR 2017-03-25 02:10:21,232 [StreamThread-2]:
> > > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - User
> > > provided listener
> > > org.apache.kafka.streams.processor.internals.StreamThread$1 for group
> > > new-part-advice failed on partition assignment
> > > java.lang.IllegalStateException: task [0_9] Log end offset of
> > > new-part-advice-key-table-changelog-9 should not change while
> restoring:
> > > old end offset 647352, current offset 647632
> > >
> > > So you can clearly see that while restoring the state some other thread
> > (on
> > > same or other instance) did commit some more offset for this change
> long
> > > partition so in the end of the process two offsets did not match. I
> think
> > > this is fairly a reasonable scenario and while restoring the state it
> > > should also consider any added offsets and not assume that this is the
> > only
> > > thread processing that partition. It may have been some other instance
> > did
> > > commit some more offsets while this thread is trying to restore the
> > state.
> > >
> > > So I feel this exception should be handled and not thrown all the way
> to
> > > the streams.
> > >
> > > What do you all think.
> > >
> > > Thanks
> > > Sachin
> > >
> >
>
Re: kafka streams 0.10.2 failing while restoring offsets
Posted by Damian Guy <da...@gmail.com>.
Yes, but we don't know why it is still processing the data. We don't want
to have multiple processes acting on the same tasks, hence the exception.
What if for some reason the other task is processing a large backlog, how
long do we wait before we give up?
I think in this case the exception is the right thing to do
On Mon, 27 Mar 2017 at 09:24 Sachin Mittal <sj...@gmail.com> wrote:
Hi,
These are the logs
https://www.dropbox.com/s/2t4ysfdqbtmcusq/complete_84_85_87_log.zip?dl=0
I think this may not always be the case especially if previous owner is on
a different machine.
Say it is processing and it takes more than the poll timeout to process or
commit the offset.
The group bumps this thread and assigns its task to a different thread on
maybe a different machine.
All this while this client may be pushing the changelog data and other
thread restoring the state from the same partition.
So between the time it starts and it seeks to the end of the changelog it
may be possible that previous thread which was still in process since it
did not know it got bumped out added some more data to that changelog.
Only when previous thread tries to commit the offset it gets to know that
it is no longer the owner of the partition and then issues a rejoin request.
I think in such a case should be handled within streams application.
Thanks
Sachin
On Mon, Mar 27, 2017 at 1:25 PM, Damian Guy <da...@gmail.com> wrote:
> Hi Sachin,
>
> This should not happen. The previous owner of the task should have stopped
> processing before the restoration begins. So if this is happening, then
> that signals a bug. Do you have more logs?
>
> Thanks,
> Damian
>
> On Sat, 25 Mar 2017 at 08:20 Sachin Mittal <sj...@gmail.com> wrote:
>
> > Hi,
> > So recently we fixed the deadlock issue and also built the streams jar
by
> > copying the rocks db configs from trunk.
> > So we don't get any deadlock issue now and also we see that the wait
time
> > of CPU cores stays around 5% (down from 50% earlier).
> >
> > However we now get a new exception which is not handled by streams
> > application and causes the instance to shutdown.
> >
> > org.apache.kafka.streams.errors.StreamsException: stream-thread
> > [StreamThread-2] Failed to rebalance
> > at
> >
> > org.apache.kafka.streams.processor.internals.StreamThread.runLoop(
> StreamThread.java:622)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.
> StreamThread.run(StreamThread.java:378)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > Caused by: java.lang.IllegalStateException: task [0_9] Log end offset of
> > new-part-advice-key-table-changelog-9 should not change while restoring:
> > old end offset 647352, current offset 647632
> > at
> >
> > org.apache.kafka.streams.processor.internals.ProcessorStateManager.
> restoreActiveState(ProcessorStateManager.java:240)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.ProcessorStateManager.
> register(ProcessorStateManager.java:193)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.AbstractProcessorContext.
> register(AbstractProcessorContext.java:99)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.
> init(RocksDBSegmentedBytesStore.java:101)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesSto
> re.init(ChangeLoggingSegmentedBytesStore.java:68)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.
> init(MeteredSegmentedBytesStore.java:66)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.state.internals.RocksDBWindowStore.init(
> RocksDBWindowStore.java:76)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.AbstractTask.
> initializeStateStores(AbstractTask.java:86)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.
> StreamTask.<init>(StreamTask.java:141)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> >
> > What I check from logs is this
> > DEBUG 2017-03-25 02:07:24,499 [StreamThread-2]:
> > org.apache.kafka.streams.processor.internals.StreamThread -
> stream-thread
> > [StreamThread-2] creating new task 0_9
> > So it creates the task at this time.
> >
> > To create the local state store from the chnagelog topic it starts at
> >
> > DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.KafkaConsumer - Subscribed to
> > partition(s): new-part-advice-key-table-changelog-9
> > DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to end of
> > partition new-part-advice-key-table-changelog-9
> > DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset
> for
> > partition new-part-advice-key-table-changelog-9 to latest offset.
> > DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.Fetcher - Handling
> > ListOffsetResponse response for new-part-advice-key-table-changelog-9.
> > Fetched offset 647352, timestamp -1
> > DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to beginning
> of
> > partition new-part-advice-key-table-changelog-9
> > DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset
> for
> > partition new-part-advice-key-table-changelog-9 to earliest offset.
> >
> > and process is over at
> > DEBUG 2017-03-25 02:10:21,225 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.Fetcher - Sending fetch for
> > partitions [new-part-advice-key-table-changelog-9] to broker
> > 192.168.73.199:9092 (id: 5 rack: null)
> > DEBUG 2017-03-25 02:10:21,230 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.KafkaConsumer - Unsubscribed all
> topics
> > or patterns and assigned partitions
> >
> > And the exception is thrown at:
> > ERROR 2017-03-25 02:10:21,232 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - User
> > provided listener
> > org.apache.kafka.streams.processor.internals.StreamThread$1 for group
> > new-part-advice failed on partition assignment
> > java.lang.IllegalStateException: task [0_9] Log end offset of
> > new-part-advice-key-table-changelog-9 should not change while restoring:
> > old end offset 647352, current offset 647632
> >
> > So you can clearly see that while restoring the state some other thread
> (on
> > same or other instance) did commit some more offset for this change long
> > partition so in the end of the process two offsets did not match. I
think
> > this is fairly a reasonable scenario and while restoring the state it
> > should also consider any added offsets and not assume that this is the
> only
> > thread processing that partition. It may have been some other instance
> did
> > commit some more offsets while this thread is trying to restore the
> state.
> >
> > So I feel this exception should be handled and not thrown all the way to
> > the streams.
> >
> > What do you all think.
> >
> > Thanks
> > Sachin
> >
>
Re: kafka streams 0.10.2 failing while restoring offsets
Posted by Sachin Mittal <sj...@gmail.com>.
Hi,
These are the logs
https://www.dropbox.com/s/2t4ysfdqbtmcusq/complete_84_85_87_log.zip?dl=0
I think this may not always be the case especially if previous owner is on
a different machine.
Say it is processing and it takes more than the poll timeout to process or
commit the offset.
The group bumps this thread and assigns its task to a different thread on
maybe a different machine.
All this while this client may be pushing the changelog data and other
thread restoring the state from the same partition.
So between the time it starts and it seeks to the end of the changelog it
may be possible that previous thread which was still in process since it
did not know it got bumped out added some more data to that changelog.
Only when previous thread tries to commit the offset it gets to know that
it is no longer the owner of the partition and then issues a rejoin request.
I think in such a case should be handled within streams application.
Thanks
Sachin
On Mon, Mar 27, 2017 at 1:25 PM, Damian Guy <da...@gmail.com> wrote:
> Hi Sachin,
>
> This should not happen. The previous owner of the task should have stopped
> processing before the restoration begins. So if this is happening, then
> that signals a bug. Do you have more logs?
>
> Thanks,
> Damian
>
> On Sat, 25 Mar 2017 at 08:20 Sachin Mittal <sj...@gmail.com> wrote:
>
> > Hi,
> > So recently we fixed the deadlock issue and also built the streams jar by
> > copying the rocks db configs from trunk.
> > So we don't get any deadlock issue now and also we see that the wait time
> > of CPU cores stays around 5% (down from 50% earlier).
> >
> > However we now get a new exception which is not handled by streams
> > application and causes the instance to shutdown.
> >
> > org.apache.kafka.streams.errors.StreamsException: stream-thread
> > [StreamThread-2] Failed to rebalance
> > at
> >
> > org.apache.kafka.streams.processor.internals.StreamThread.runLoop(
> StreamThread.java:622)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.
> StreamThread.run(StreamThread.java:378)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > Caused by: java.lang.IllegalStateException: task [0_9] Log end offset of
> > new-part-advice-key-table-changelog-9 should not change while restoring:
> > old end offset 647352, current offset 647632
> > at
> >
> > org.apache.kafka.streams.processor.internals.ProcessorStateManager.
> restoreActiveState(ProcessorStateManager.java:240)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.ProcessorStateManager.
> register(ProcessorStateManager.java:193)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.AbstractProcessorContext.
> register(AbstractProcessorContext.java:99)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.
> init(RocksDBSegmentedBytesStore.java:101)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesSto
> re.init(ChangeLoggingSegmentedBytesStore.java:68)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.
> init(MeteredSegmentedBytesStore.java:66)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.state.internals.RocksDBWindowStore.init(
> RocksDBWindowStore.java:76)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.AbstractTask.
> initializeStateStores(AbstractTask.java:86)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> > at
> >
> > org.apache.kafka.streams.processor.internals.
> StreamTask.<init>(StreamTask.java:141)
> > ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> >
> > What I check from logs is this
> > DEBUG 2017-03-25 02:07:24,499 [StreamThread-2]:
> > org.apache.kafka.streams.processor.internals.StreamThread -
> stream-thread
> > [StreamThread-2] creating new task 0_9
> > So it creates the task at this time.
> >
> > To create the local state store from the chnagelog topic it starts at
> >
> > DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.KafkaConsumer - Subscribed to
> > partition(s): new-part-advice-key-table-changelog-9
> > DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to end of
> > partition new-part-advice-key-table-changelog-9
> > DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset
> for
> > partition new-part-advice-key-table-changelog-9 to latest offset.
> > DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.Fetcher - Handling
> > ListOffsetResponse response for new-part-advice-key-table-changelog-9.
> > Fetched offset 647352, timestamp -1
> > DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to beginning
> of
> > partition new-part-advice-key-table-changelog-9
> > DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset
> for
> > partition new-part-advice-key-table-changelog-9 to earliest offset.
> >
> > and process is over at
> > DEBUG 2017-03-25 02:10:21,225 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.Fetcher - Sending fetch for
> > partitions [new-part-advice-key-table-changelog-9] to broker
> > 192.168.73.199:9092 (id: 5 rack: null)
> > DEBUG 2017-03-25 02:10:21,230 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.KafkaConsumer - Unsubscribed all
> topics
> > or patterns and assigned partitions
> >
> > And the exception is thrown at:
> > ERROR 2017-03-25 02:10:21,232 [StreamThread-2]:
> > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - User
> > provided listener
> > org.apache.kafka.streams.processor.internals.StreamThread$1 for group
> > new-part-advice failed on partition assignment
> > java.lang.IllegalStateException: task [0_9] Log end offset of
> > new-part-advice-key-table-changelog-9 should not change while restoring:
> > old end offset 647352, current offset 647632
> >
> > So you can clearly see that while restoring the state some other thread
> (on
> > same or other instance) did commit some more offset for this change long
> > partition so in the end of the process two offsets did not match. I think
> > this is fairly a reasonable scenario and while restoring the state it
> > should also consider any added offsets and not assume that this is the
> only
> > thread processing that partition. It may have been some other instance
> did
> > commit some more offsets while this thread is trying to restore the
> state.
> >
> > So I feel this exception should be handled and not thrown all the way to
> > the streams.
> >
> > What do you all think.
> >
> > Thanks
> > Sachin
> >
>
Re: kafka streams 0.10.2 failing while restoring offsets
Posted by Damian Guy <da...@gmail.com>.
Hi Sachin,
This should not happen. The previous owner of the task should have stopped
processing before the restoration begins. So if this is happening, then
that signals a bug. Do you have more logs?
Thanks,
Damian
On Sat, 25 Mar 2017 at 08:20 Sachin Mittal <sj...@gmail.com> wrote:
> Hi,
> So recently we fixed the deadlock issue and also built the streams jar by
> copying the rocks db configs from trunk.
> So we don't get any deadlock issue now and also we see that the wait time
> of CPU cores stays around 5% (down from 50% earlier).
>
> However we now get a new exception which is not handled by streams
> application and causes the instance to shutdown.
>
> org.apache.kafka.streams.errors.StreamsException: stream-thread
> [StreamThread-2] Failed to rebalance
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:622)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:378)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> Caused by: java.lang.IllegalStateException: task [0_9] Log end offset of
> new-part-advice-key-table-changelog-9 should not change while restoring:
> old end offset 647352, current offset 647632
> at
>
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.restoreActiveState(ProcessorStateManager.java:240)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> at
>
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:193)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> at
>
> org.apache.kafka.streams.processor.internals.AbstractProcessorContext.register(AbstractProcessorContext.java:99)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> at
>
> org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.init(RocksDBSegmentedBytesStore.java:101)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> at
>
> org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.init(ChangeLoggingSegmentedBytesStore.java:68)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> at
>
> org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.init(MeteredSegmentedBytesStore.java:66)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> at
>
> org.apache.kafka.streams.state.internals.RocksDBWindowStore.init(RocksDBWindowStore.java:76)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> at
>
> org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:86)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
> at
>
> org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:141)
> ~[kafka-streams-0.10.2.1-SNAPSHOT.jar:na]
>
> What I check from logs is this
> DEBUG 2017-03-25 02:07:24,499 [StreamThread-2]:
> org.apache.kafka.streams.processor.internals.StreamThread - stream-thread
> [StreamThread-2] creating new task 0_9
> So it creates the task at this time.
>
> To create the local state store from the chnagelog topic it starts at
>
> DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> org.apache.kafka.clients.consumer.KafkaConsumer - Subscribed to
> partition(s): new-part-advice-key-table-changelog-9
> DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to end of
> partition new-part-advice-key-table-changelog-9
> DEBUG 2017-03-25 02:07:24,550 [StreamThread-2]:
> org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset for
> partition new-part-advice-key-table-changelog-9 to latest offset.
> DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> org.apache.kafka.clients.consumer.internals.Fetcher - Handling
> ListOffsetResponse response for new-part-advice-key-table-changelog-9.
> Fetched offset 647352, timestamp -1
> DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> org.apache.kafka.clients.consumer.KafkaConsumer - Seeking to beginning of
> partition new-part-advice-key-table-changelog-9
> DEBUG 2017-03-25 02:07:24,552 [StreamThread-2]:
> org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset for
> partition new-part-advice-key-table-changelog-9 to earliest offset.
>
> and process is over at
> DEBUG 2017-03-25 02:10:21,225 [StreamThread-2]:
> org.apache.kafka.clients.consumer.internals.Fetcher - Sending fetch for
> partitions [new-part-advice-key-table-changelog-9] to broker
> 192.168.73.199:9092 (id: 5 rack: null)
> DEBUG 2017-03-25 02:10:21,230 [StreamThread-2]:
> org.apache.kafka.clients.consumer.KafkaConsumer - Unsubscribed all topics
> or patterns and assigned partitions
>
> And the exception is thrown at:
> ERROR 2017-03-25 02:10:21,232 [StreamThread-2]:
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - User
> provided listener
> org.apache.kafka.streams.processor.internals.StreamThread$1 for group
> new-part-advice failed on partition assignment
> java.lang.IllegalStateException: task [0_9] Log end offset of
> new-part-advice-key-table-changelog-9 should not change while restoring:
> old end offset 647352, current offset 647632
>
> So you can clearly see that while restoring the state some other thread (on
> same or other instance) did commit some more offset for this change long
> partition so in the end of the process two offsets did not match. I think
> this is fairly a reasonable scenario and while restoring the state it
> should also consider any added offsets and not assume that this is the only
> thread processing that partition. It may have been some other instance did
> commit some more offsets while this thread is trying to restore the state.
>
> So I feel this exception should be handled and not thrown all the way to
> the streams.
>
> What do you all think.
>
> Thanks
> Sachin
>