SAMZA-2317: ProcessJob does not call CoordinatorStreamStore.close()#1289
Merged
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 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: Users when deploying a job in dev will see an error message from the kafka consumer finalizer complaining about "kafka consumer allocated and not closed"
This error message will appear multiple times in their job and doesn't actually affect the job's performance or correctness. This leads to users falsely believing that any job failures is because of this message.
Cause:
All Kafka consumers have a finalize method in them. This means when the garbage collector determines that there's no more references to the consumer, it will try to close the consumer and find that the consumer wasn't closed properly yet. This happens because in the ProcessJobFactory, a CoordinatorStreamStore is initialized. It is then wrapped around a NamespaceAwareCoordinatorStreamStore. Later on when the ProcessJob is done, a close() is called on the NamespaceAwareCoordinatorStreamStore, however this doesn't close the original CoordinatorStreamStore (As there could be multiple NamespaceAware stores pointing to one CoordinatorStreamStore). So, from the garbage collector's perspective, it will see that there are no more references to the CoordinatorStreamStore, leading to the finalize method reporting the above error.
Changes: ProcessJob will now have an additional parameter which will be the CoordinatorStreamStore. Then a close is called on the store when the job is done running. That way there is still a reference to the store after the ProcessJob starts. Also updated the TestProcessJob file to test that the store is closed properly.
This way of fixing this error was chosen as any other solution would involved changing how the coordinatorStreamStore operates which could lead to many unintended side-effects
Tests: To reproduce this error, deploy any job running the latest version of Samza and look in the .out log file. Then run the same job with this fix included and that error message shouldn't appear anymore. Tested this with a test a build in LinkedIn internal to verify the error message isn't there anymore. Also ran "./gradlew clean build" and "./gradlew clean test"