New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[FLINK-10774] Rework lifecycle management of partitionDiscoverer in FlinkKafkaConsumerBase #7606
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Since I've already reviewed this as part of the efforts in #7020, and Travis is green, LGTM +1 👍
Thanks for fixing this @stevenzwu @tillrohrmann
Thanks for the review @tzulitai. Merging. |
…linkKafkaConsumerBase
…cle verifications
…before closed when concurrently accessed
…LifeCycle Split #testConsumerLifeCycle into two methods which represent the two if-else branches. This closes apache#7606.
afea47e
to
2fd4549
Compare
…LifeCycle Split #testConsumerLifeCycle into two methods which represent the two if-else branches. This closes apache#7606.
…LifeCycle Split #testConsumerLifeCycle into two methods which represent the two if-else branches. This closes apache#7606.
…LifeCycle Split #testConsumerLifeCycle into two methods which represent the two if-else branches. This closes apache#7606.
Failing test case is unrelated. |
@tillrohrmann sorry for a late question. you mentioned In production/distributed environment, when a job restart due to exception from |
What is the purpose of the change
This PR reworks the lifecycle management of the
FlinkKafkaConsumerBase#partitionDiscoverer
such that it will always be closed in theclose()
method. This implies that all users of this class need to callclose()
in order to properly clean up resources.cc @stevenzwu @tzulitai
Brief change log
partitionDiscoverer
from#run
methodFlinkKafkaConsumerBase#cancel
and#close
such that cancel is non blocking andclose
will release all required resourcesVerifying this change
FlinkKafkaConsumerBaseTest#testClosePartitionDiscovererWhenOpenThrowException
,#testClosePartitionDiscovererWhenCreateKafkaFetcherFails
,#testClosePartitionDiscovererWhenKafkaFetcherFails
and#testClosePartitionDiscovererWithCancellation
Does this pull request potentially affect one of the following parts:
@Public(Evolving)
: (no)Documentation