SAMZA-2305: Stream processor should ensure previous container is stopped during a rebalance #1213
Conversation
This file contains hidden or 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
mynameborat
changed the title
dnishimura
suggested changes
Nov 7, 2019
Contributor
dnishimura
left a comment
dnishimura
left a comment
There was a problem hiding this comment.
Thanks for this important fix! Just a few comments and questions.
samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
Outdated
Show resolved
Hide resolved
samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
Outdated
Show resolved
Hide resolved
dnishimura
approved these changes
Nov 13, 2019
Contributor
dnishimura
left a comment
dnishimura
left a comment
There was a problem hiding this comment.
LGTM! Thanks for the fix.
mynameborat
changed the title
sborya
reviewed
Nov 18, 2019
samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
Outdated
Show resolved
Hide resolved
samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
Show resolved
Hide resolved
samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
Show resolved
Hide resolved
sborya
approved these changes
Nov 20, 2019
Contributor
cameronlee314
left a comment
cameronlee314
left a comment
There was a problem hiding this comment.
My overall feedback as a separate reader:
It seems like using InterruptedExceptions is kind of an "implicit" way of handling this case, so it's a bit hard to validate that all places have been updated correctly (and might be a bit hard to follow if anyone else has to update this flow in the future). I see that you have done good analysis on alternative options, so I think it is reasonable to stick with this strategy overall, but if there is anything further you can think of to make it as explicit as possible, it might be helpful. You have already added some good things (i.e. update method sig with throws InterruptedException, javadocs), but I'm not sure if there is anything else you can do.
samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
Show resolved
Hide resolved
samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
Outdated
Show resolved
Hide resolved
samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
Show resolved
Hide resolved
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java
Show resolved
Hide resolved
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java
Show resolved
Hide resolved
samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java
Outdated
Show resolved
Hide resolved
mynameborat
force-pushed
the
SAMZA-2305
branch
from
59530a4 to
15a56bf
Compare
Contributor
cameronlee314
left a comment
cameronlee314
left a comment
There was a problem hiding this comment.
Please update the description notes to reflect the API change for the InterruptedException.
samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngine.scala
Show resolved
Hide resolved
cameronlee314
approved these changes
Dec 10, 2019
samza-kv/src/test/scala/org/apache/samza/storage/kv/TestKeyValueStorageEngine.scala
Show resolved
Hide resolved
samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngine.scala
Show resolved
Hide resolved
cameronlee314
approved these changes
Dec 10, 2019
rmatharu-zz
pushed a commit
to rmatharu-zz/samza
that referenced
this pull request
Jan 21, 2020
…ped during a rebalance (apache#1213) Stream processor should ensure previous container is stopped during a rebalance
lhaiesp
pushed a commit
that referenced
this pull request
Feb 13, 2020
…ped during a rebalance (#1213) Stream processor should ensure previous container is stopped during a rebalance
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
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.
Symptom: Duplicate processing, Inconsistent checkpoints for inputs, Inconsistent changelog state
Cause: We have a bug in the state machine inside stream processor that can result in processors running containers with old job model version after rebalances in Standalone deployment model.
Fix: We interrupt the container and wait for container to shutdown gracefully within a timeout (task.shutdown.ms) and fail the stream processor if the container doesn’t shut down within the timeout
Tests: Added unit tests for StreamProcessor and SamzaContainer. Working on integration tests.
API Changes
restoremethods onTaskRestoreManagerandStorageEnginenow throwsInterruptedException. Please refer to java docs to get additional implementation notes.Upgrade Instructions Standalone jobs w/ external monitoring service to restart your application should follow the External monitoring section in the document below to tune debounce time to account for monitoring service latency and container startup time.
Usage Instructions None
More details about the bug can be found here.