Please find the JIRA https://issues.apache.org/jira/browse/KAFKA-4848
On Mon, Mar 6, 2017 at 5:20 PM, Damian Guy <damian....@gmail.com> wrote: > Hi Sachin, > > If it is a bug then please file a JIRA for it, too. > > Thanks, > Damian > > On Mon, 6 Mar 2017 at 11:23 Sachin Mittal <sjmit...@gmail.com> wrote: > > > Ok that's great. > > So you have already fixed that issue. > > > > I have modified my PR to remove that change (which was done keeping > > 0.10.2.0 in mind). > > > > However the other issue is still valid. > > > > Please review that change. https://github.com/apache/kafka/pull/2642 > > > > > > Thanks > > Sachin > > > > > > On Mon, Mar 6, 2017 at 3:56 PM, Damian Guy <damian....@gmail.com> wrote: > > > > > On trunk the CommitFailedException isn't thrown anymore. The > > commitOffsets > > > method doesn't throw an exception. It returns one if it was thrown. We > > used > > > to throw this exception during suspendTasksAndState, but we don't > > anymore. > > > > > > On Mon, 6 Mar 2017 at 05:04 Sachin Mittal <sjmit...@gmail.com> wrote: > > > > > > > Hi > > > > On CommitFailedException at onPartitionsRevoked if it is thrown it > gets > > > > assigned to rebalanceException. > > > > This causes the stream thread to shutdown. I am not sure how we can > > > resume > > > > the thread. > > > > > > > > Note thread is not in invalid state because because it already has > been > > > > assigned new partitions and this exception happens when trying to > > revoke > > > > old partitions which have been moved to some other thread, so we need > > to > > > > swallow this exception at the StreanThread side too, just like we > > swallow > > > > it at ConsumerCoordinator.java > > > > > > > > Also I fixed this against code base 0.10.2.0 and the difference in > that > > > vs > > > > trunk code is these lines > > > > 10.2.0 > > > > if (firstException.get() == null) { > > > > firstException.set(commitOffsets()); > > > > } > > > > vs trunk > > > > if (firstException.get() == null) { > > > > // TODO: currently commit failures will not be thrown to > > > users > > > > // while suspending tasks; this need to be re-visit after > > > > KIP-98 > > > > commitOffsets(); > > > > } > > > > I am again not sure since this part is still a TODO, but looking at > > code > > > I > > > > see that commitOffsets can still throw the CommitFailedException > which > > > > needs to be handled at onPartitionsRevoked. > > > > > > > > Hope this makes sense. > > > > > > > > On second issue, the deadlock is not caused by CommitFailedExceptio, > > but > > > > after fixing the deadlock we need to make sure thread does not die > due > > to > > > > unhandled CommitFailedException at onPartitionsRevoked. > > > > The deadlock issue is like this. > > > > If a thread has two partitions and while processing partition one it > > > takes > > > > more than MAX_POLL_INTERVAL_MS_CONFIG time, then this thread is > evicted > > > > from the group and both partitions are now migrated to some other > > thread. > > > > Now when it tries to process the partition two it tries to get the > lock > > > to > > > > rocks db. It won't get the lock since that partition is now moved to > > some > > > > other thread. So it keeps increasing the backoffTimeMs and keeps > trying > > > to > > > > get the lock forever. This reaching a deadlock. > > > > To fix this we need some upper bound of the time limit till it tries > to > > > get > > > > that lock. And that upper bound has to be > MAX_POLL_INTERVAL_MS_CONFIG, > > > > because if by that time it has not got the lock, we can see that this > > > > thread was evicted from the group and need to rejoin again to get new > > > > partitions. > > > > > > > > On JIRA issue I can create one and attach the part of logs where it > > keeps > > > > trying to get the lock with increasing backoffTimeM. > > > > > > > > Let me know if these makes sense. Right now this is the best way we > > could > > > > come up with to handle stream thread failures. > > > > > > > > Also on a side note I feel we need more resilient streams. If we have > > say > > > > configured our streams application with 4 threads and for whatever > > > reason a > > > > thread dies, then application should itself (or via some exposed > > hooks), > > > > allow to restart a new thread (because in Java I guess same thread > > cannot > > > > be restarted), so that number of threads always stay what one has > > > > configured. > > > > I think exposed hooks will be better option to do this. > > > > > > > > Thanks > > > > Sachin > > > > > > > > > > > > > > > > > > > > On Mon, Mar 6, 2017 at 5:40 AM, Matthias J. Sax < > matth...@confluent.io > > > > > > > wrote: > > > > > > > > > Sachin, > > > > > > > > > > thanks a lot for contributing! > > > > > > > > > > Right now, I am not sure if I understand the change. On > > > > > CommitFailedException, why can we just resume the thread? To me, it > > > > > seems that the thread will be in an invalid state and thus it's not > > > save > > > > > to just swallow the exception and keep going. Can you shed some > > light? > > > > > > > > > > And from my understanding, the deadlock is "caused" by the change > > from > > > > > above, right? So if it is save to swallow the exception, we should > do > > > > > some "clean up" to avoid the deadlock in the first place, instead > of > > > > > applying and additional timeout. > > > > > > > > > > Also, if this is a bug, we should have a JIRA. > > > > > > > > > > -Matthias > > > > > > > > > > > > > > > On 3/5/17 4:11 AM, Sachin Mittal wrote: > > > > > > Hi, > > > > > > Please find the new PR > > > > > > https://github.com/apache/kafka/pull/2642/ > > > > > > > > > > > > I see that in trunk there has been change which is different from > > in > > > > > 10.2.0 > > > > > > > > > > > > 10.2.0 > > > > > > if (firstException.get() == null) { > > > > > > firstException.set(commitOffsets()); > > > > > > } > > > > > > vs trunk > > > > > > if (firstException.get() == null) { > > > > > > // TODO: currently commit failures will not be thrown > > to > > > > > users > > > > > > // while suspending tasks; this need to be re-visit > > after > > > > > KIP-98 > > > > > > commitOffsets(); > > > > > > } > > > > > > I am not sure in view of this is is my part of the fix still > valid. > > > > Looks > > > > > > like it is still valid. > > > > > > > > > > > > Also on side note what is the policy of closing a branch that is > > just > > > > > > released. > > > > > > > > > > > > Since you have release 10.2.0 we are using that and that is why > > have > > > > made > > > > > > changes in that branch so that our changes just modify the needed > > > code > > > > > and > > > > > > we don't mess up the other released code. > > > > > > > > > > > > Is the new release released off the branch 10.2.0, if yes then > you > > > > should > > > > > > not close it as there can be patch fixes on them. > > > > > > > > > > > > Or is the release always made off the branch trunk. In that case > > how > > > > can > > > > > we > > > > > > pick up the code on which the release binaries were created so > when > > > we > > > > > > build the binary we have exactly same code as released one, plus > > any > > > > > > changes (we or someone else) makes on it. > > > > > > > > > > > > Also if a branch is closed, then perhaps we should delete it or > > mark > > > it > > > > > > closed or something. > > > > > > > > > > > > Please let us know how releases get created (off what codebase), > so > > > we > > > > > are > > > > > > more exact in applying our changes to. > > > > > > > > > > > > Thanks > > > > > > Sachin > > > > > > > > > > > > > > > > > > > > > > > > On Sun, Mar 5, 2017 at 3:40 PM, Eno Thereska < > > eno.there...@gmail.com > > > > > > > > > wrote: > > > > > > > > > > > >> Thanks Sachin, one thing before the review, 0.10.2 is closed > now, > > > this > > > > > >> needs to target trunk. > > > > > >> > > > > > >> Thanks > > > > > >> Eno > > > > > >>> On 5 Mar 2017, at 09:10, Sachin Mittal <sjmit...@gmail.com> > > wrote: > > > > > >>> > > > > > >>> Please review the PR and let me know if this makes sense. > > > > > >>> > > > > > >>> https://github.com/apache/kafka/pull/2640 > > > > > >>> > > > > > >>> Thanks > > > > > >>> Sachin > > > > > >>> > > > > > >>> > > > > > >>> On Sun, Mar 5, 2017 at 1:49 PM, Eno Thereska < > > > eno.there...@gmail.com > > > > > > > > > > >> wrote: > > > > > >>> > > > > > >>>> Thanks Sachin for your contribution. Could you create a pull > > > request > > > > > out > > > > > >>>> of the commit (so we can add comments, and also so you are > > > > > acknowledged > > > > > >>>> properly for your contribution)? > > > > > >>>> > > > > > >>>> Thanks > > > > > >>>> Eno > > > > > >>>>> On 5 Mar 2017, at 07:34, Sachin Mittal <sjmit...@gmail.com> > > > wrote: > > > > > >>>>> > > > > > >>>>> Hi, > > > > > >>>>> So far in our experiment we have encountered 2 critical bugs. > > > > > >>>>> 1. If a thread takes more that MAX_POLL_INTERVAL_MS_CONFIG to > > > > > compute a > > > > > >>>>> cycle it gets evicted from group and rebalance takes place > and > > it > > > > > gets > > > > > >>>> new > > > > > >>>>> assignment. > > > > > >>>>> However when this thread tries to commit offsets for the > > revoked > > > > > >>>> partitions > > > > > >>>>> in > > > > > >>>>> onPartitionsRevoked it will again throw the > > > CommitFailedException. > > > > > >>>>> > > > > > >>>>> This gets handled by ConsumerCoordinatorso there is no point > to > > > > > assign > > > > > >>>> this > > > > > >>>>> exception to > > > > > >>>>> rebalanceException in StreamThread and stop it. It has > already > > > been > > > > > >>>>> assigned new partitions and it can continue. > > > > > >>>>> > > > > > >>>>> So as fix in case on CommitFailedException I am not killing > the > > > > > >>>> StreamThrea. > > > > > >>>>> > > > > > >>>>> 2. Next we see a deadlock state when to process a task it > takes > > > > > longer > > > > > >>>>> than MAX_POLL_INTERVAL_MS_CONFIG > > > > > >>>>> time. Then this threads partitions are assigned to some other > > > > thread > > > > > >>>>> including rocksdb lock. When it tries to process the next > task > > it > > > > > >> cannot > > > > > >>>>> get rocks db lock and simply keeps waiting for that lock > > forever. > > > > > >>>>> > > > > > >>>>> in retryWithBackoff for AbstractTaskCreator we have a > > > > backoffTimeMs = > > > > > >>>> 50L. > > > > > >>>>> If it does not get lock the we simply increase the time by > 10x > > > and > > > > > keep > > > > > >>>>> trying inside the while true loop. > > > > > >>>>> > > > > > >>>>> We need to have a upper bound for this backoffTimeM. If the > > time > > > is > > > > > >>>> greater > > > > > >>>>> than MAX_POLL_INTERVAL_MS_CONFIG and it still hasn't got the > > > lock > > > > > >> means > > > > > >>>>> this thread's partitions are moved somewhere else and it may > > not > > > > get > > > > > >> the > > > > > >>>>> lock again. > > > > > >>>>> > > > > > >>>>> So I have added an upper bound check in that while loop. > > > > > >>>>> > > > > > >>>>> The commits are here: > > > > > >>>>> https://github.com/sjmittal/kafka/commit/ > > > > > >> 6f04327c890c58cab9b1ae108af4ce > > > > > >>>> 5c4e3b89a1 > > > > > >>>>> > > > > > >>>>> please review and if you feel they make sense, please merge > it > > to > > > > > main > > > > > >>>>> branch. > > > > > >>>>> > > > > > >>>>> Thanks > > > > > >>>>> Sachin > > > > > >>>> > > > > > >>>> > > > > > >> > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > >
