Shutdown Kafka consumers and producer when the external stream is stopped #645
Conversation
| std::string name() const { return rd_kafka_name(rk.get()); } | ||
|
|
||
| private: | ||
| void backgroundPoll(UInt64 poll_timeout_ms) const; | ||
|
|
||
| klog::KafkaPtr rk {nullptr, rd_kafka_destroy}; | ||
| ThreadPool poller; | ||
| std::atomic_flag stopped; | ||
| bool stopped {false}; |
There was a problem hiding this comment.
multiple thread unsafe ? one thread call shutdown, another thread call consumeBatch
There was a problem hiding this comment.
It appears there is still race
There was a problem hiding this comment.
No, an atomic is not needed here, there won't be a race.
So, the stopped flag is to for telling the Consumer (and the Producer in the below case) that it's already stopped. And, when we set stopped to true, we don't do anything else ( like, we don't detroy anything ), so there is nothing to guard with it, i.e. there is no such logic in our code:
if (!stopped)
{
stopped = true;
/// do something
}
else
{
/// do something else
}So there is no race.
The purpose is to to let operations on the Consumer fail, but it does not have to fail immediately. Take your case as an example, if one thread call shutdown, and another call consumeBatch, it's fine, it does not care which one wins, even consumeBatch wins, nothing bad will happen, and next time when consumeBatch is called, it will fail as expected, and then, KafkaSource will stopped reading more data. And also, we don't even need to guard shutdown, it can be called multiple times, no side effects.
Please let me know if this does not make sense to you.
There was a problem hiding this comment.
Regardless it does not have any impact in current implementation, I reverted it back, and now use std::atomic_flag.
There was a problem hiding this comment.
The race i meant it not the stopped data member but at high level..
- Client call
isStopped(), returns false, context switch to 2) - Consumer then stopped, switch back to 1)
- Client already checked consumer is not stopped, but it actually is, continue operating on stopped consumer ?
There was a problem hiding this comment.
Same for producer and we run into this problem before
| std::string name() const { return rd_kafka_name(rk.get()); } | ||
|
|
||
| private: | ||
| void backgroundPoll(UInt64 poll_timeout_ms) const; | ||
|
|
||
| klog::KafkaPtr rk {nullptr, rd_kafka_destroy}; | ||
| ThreadPool poller; | ||
| std::atomic_flag stopped; | ||
| bool stopped {false}; |
There was a problem hiding this comment.
It appears there is still race
zliang-min
force-pushed
the
kafka/graceful-shutdown-2
branch
from
5846533 to
c0511fe
Compare
|
@chenziliang @yl-lisen I reverted the change and use |
jovezhong
changed the title
|
(Jove Github Bot) added it to the current sprint. |
|
(Jove Github Bot) moved this ticket out of the GitHub project(up to 1200 tickets for one project). |
PR checklist:
proton: starts/endsfor new code in existing community code base ?Please write user-readable short description of the changes:
A follow-up change of #643. In this change, it makes sure that if an external stream is stopped, all consumers and producer of that stream will be shutdown, and all ad-hoc queries will fail.