KAFKA-6844: Call shutdown on GlobalStreamThread after all StreamThreads have stopped #4950
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
bbejeck
changed the title
bbejeck
changed the title
tedyu
approved these changes
May 1, 2018
mjsax
approved these changes
May 2, 2018
|
@mjsax I had similar thoughts, I just ran out of time yesterday. I'll take look at getting a test together. |
|
LGTM! |
|
added an integration test |
guozhangwang
reviewed
May 4, 2018
| @Override | ||
| public void process(final String key, final Long value) { | ||
| // need to simulate slow processing | ||
| Utils.sleep(1000); |
There was a problem hiding this comment.
I think a better test scenario is to move the logic in close() call, i.e. when the stream thread is being shutdown, and topology is closing, we call processorNode.close() in which we wait for a while and then tries to access the global store. It mimics the case where in closing the store cache is flushed and hence tries to access the global store again.
|
@guozhangwang updated this |
|
@bbejeck Jenkins failure is due to a recent PR removing the deprecated API. |
bbejeck
force-pushed
the
KAFKA-6844_race_condition_between_shutting_down_stream_and_global_threads
branch
from
a09a661
to
d6606fb
Compare
|
rebased this |
ying-zheng
pushed a commit
to ying-zheng/kafka
that referenced
this pull request
Jul 6, 2018
…ds have stopped (apache#4950) Moved the shutdown of GlobalStreamThread to after all StreamThread instances have stopped. There can be a race condition where shut down is called on a StreamThread then shut down is called on a GlobalStreamThread, but if StreamThread is delayed in shutting down, the GlobalStreamThread can shutdown first. If the StreamThread tries to access a GlobalStateStore before closing the user can get an exception stating "..Store xxx is currently closed " Tested by running all current streams tests. Reviewers: Ted Yu <yuzhihong@gmail.com>, John Roesler <john@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>
bbejeck
deleted the
KAFKA-6844_race_condition_between_shutting_down_stream_and_global_threads
branch
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Moved the shutdown of
GlobalStreamThreadto after allStreamThreadinstances have stopped.There can be a race condition where shut down is called on a
StreamThreadthen shut down is called on a GlobalStreamThread, but if StreamThread is delayed in shutting down, the GlobalStreamThread can shutdown first.If the StreamThread tries to access a GlobalStateStore before closing the user can get an exception stating "..Store xxx is currently closed "
Tested by running all current streams tests.
Committer Checklist (excluded from commit message)