Enable Function Workers to use exclusive producer to write to internal topics #9275
Conversation
| .supplier(() -> { | ||
| try { | ||
| Producer<byte[]> producer = client.newProducer().topic(topic) | ||
| .accessMode(ProducerAccessMode.Exclusive) |
There was a problem hiding this comment.
Would it make sense to use WaitForExclusive here? So that the producer instance will stay pending until it becomes the leader?
There was a problem hiding this comment.
I don't think it is a good idea for "liveness" is we block forever which may cause deadlocks. While blocking, the worker might be not even be the leader anymore. I think we should keep retrying and periodically check if we are still the leader.
| try { | ||
| Producer<byte[]> producer = client.newProducer().topic(topic) | ||
| .accessMode(ProducerAccessMode.Exclusive) | ||
| .enableBatching(false) |
There was a problem hiding this comment.
Do you remember if there was any reason for not using batching here?
There was a problem hiding this comment.
I do not. We could enable it but I wouldn't expect much benefit as we are not exactly optimizing for metadata operation throughput. For debugging it is a little easier without batching as you will know exactly how many messages are in the internal topics.
jerrypeng
changed the title
|
What happens if the broker is still on an old version that does not support ExclusiveProducer mode ? |
If the broker hasn't been updated and does not support exclusive producing mode, the client can still connect and produce but exclusivity is not guaranteed |
|
Thanks. Makes sense |
jerrypeng
changed the title
jerrypeng
changed the title
jerrypeng
force-pushed
the
add_exclusive_producer_to_functions
branch
from
17cd361
to
2fc028a
Compare
jerrypeng
changed the title
jerrypeng
force-pushed
the
add_exclusive_producer_to_functions
branch
from
58062fd
to
b893b48
Compare
…l topics (apache#9275) Co-authored-by: Jerry Peng <jerryp@splunk.com>
Motivation
To maintain correctness, we need to use exclusive producer to write to the internal topics Pulsar Functions to manage metadata / state.