PIP 68: Exclusive Producer#8685
Conversation
merlimat
added
the
type/feature
0ceb670 to
e6f7c8e
Compare
rdhabalia
left a comment
rdhabalia
left a comment
There was a problem hiding this comment.
LGTM.. few minor comments.
| } | ||
|
|
||
| protected CompletableFuture<Optional<Long>> incrementTopicEpochIfNeeded(Producer producer) { | ||
| lock.writeLock().lock(); |
There was a problem hiding this comment.
can we avoid locking for normal producer usecase?
if (producers.isEmpty() && producer.getAccessMode() == Shared) {
return CompletableFuture.completedFuture(topicEpoch);
}
There was a problem hiding this comment.
I'm not sure that we can avoid the locking, and I'm not sure that it's something to worry about in the context of adding a producer.
Actually, in the specific case there's still a race condition between updating the producers map, since the update to hasExclusiveProducer and the insertion into producers map should be atomic as well.
| return future; | ||
| } | ||
|
|
||
| // case WaitForExclusive: |
There was a problem hiding this comment.
I have one suggestion. I am sure you must have thought about it but can't we rename WaitForExclusive with FailOver as it will be consistent to subscription type name and it will need no explanation and easy to understand.
There was a problem hiding this comment.
The problem I see is that it's not exactly the same thing and that can confuse people:
- The semantic of creating a
WaitForExclusiveis different because thenewProducer()....create()call is hanging until that particular producer is selected, unlike in consumers where it's created immediately but it will not receive messages. - The use case is mostly different from "failover" for reliability purpose, since you can use
WaitForExclusiveto do leader election
| checkArgument(producer.getTopic() == this); | ||
|
|
||
| if (producers.remove(producer.getProducerName(), producer)) { | ||
| handleProducerRemoved(producer); |
There was a problem hiding this comment.
shouldn't we need lock here else it may create a race condition and producer with WaitForExclusive may wait forever.
There was a problem hiding this comment.
No lock needed there so far. Other changes are needed for WaitForExclusive, it's not implemented in this PR
eolivelli
left a comment
eolivelli
left a comment
There was a problem hiding this comment.
Great work!
I left a few little suggestions
| } | ||
|
|
||
| protected CompletableFuture<Optional<Long>> incrementTopicEpochIfNeeded(Producer producer) { | ||
| lock.writeLock().lock(); |
6ef1fc1 to
c2d57bc
Compare
|
Rebased and addressed comments |
eolivelli
left a comment
eolivelli
left a comment
There was a problem hiding this comment.
LGTM
Great work
One question, with an Exclusive Producer, what happens if I delete and recreate a topic?
- create a topic
- start ex producer and write
- delete/create the same topic
- is the producer still valid?
I expect the ex producer to be fenced out.
Is this the case?
| } | ||
| } | ||
|
|
||
| private static final Random random = new Random(); |
There was a problem hiding this comment.
What about setting a fixed seed? In order to have reproducible tests execution.
There was a problem hiding this comment.
This is just to get random topic names. We don't really need it to be reproducible.
Deleting the topic will delete all the state associated with the topic. If the topic is re-created, it's effectively a different topic. All the state associated with the topic is gone: subscriptions, dedup state, fencing. The topic epoch will be reset after the deletion, so it should allow the producer. I will add specific test for this. |
f6aecc3 to
eba7bbe
Compare
* PIP 68: Exclusive Producer * Added missing enums cases in C++ * Addressed comments * Moved constant to top of file * Fix mistake in previous update * Added handling for topic deletion
4 participants
Motivation
Implementation of https://github.com/apache/pulsar/wiki/PIP-68%3A-Exclusive-Producer
This is the first part of the implementation of exclusive producer. It adds support for
ProducerAccessMode.Exclusive.The 2nd mode,
ProducerAccessMode.WaitForExclusive, will be added in separate PR.