Fix a couple of bugs in exclusive producer #9554
Conversation
| StringBuilder errorMsg = new StringBuilder(); | ||
| errorMsg.append("Topic has an existing exclusive producer"); | ||
| if (producers.keys().hasMoreElements()) { | ||
| errorMsg.append(": " + producers.keys().nextElement()); |
There was a problem hiding this comment.
I'm not sure this avoids the race altogether since there's no atomicity between checking hasMoreElements() and nextElement(). I think we could get an iterator and use that to check, since it will be guaranteed to be consistent.
| if (producers.keys().hasMoreElements()) { | ||
| errorMsg.append(": " + producers.keys().nextElement()); | ||
| } |
There was a problem hiding this comment.
| if (producers.keys().hasMoreElements()) { | |
| errorMsg.append(": " + producers.keys().nextElement()); | |
| } | |
| producers.keySet().stream().findFirst().map(p -> errorMsg.append(": " ).append(p)); |
| if (producers.keys().hasMoreElements()) { | ||
| errorMsg.append(": " + producers.keys().nextElement()); | ||
| } |
There was a problem hiding this comment.
| if (producers.keys().hasMoreElements()) { | |
| errorMsg.append(": " + producers.keys().nextElement()); | |
| } | |
| producers.keySet().stream().findFirst().map(p -> errorMsg.append(": " ).append(p)); |
1. Fix uncaught exception in exclusive producer 2. Incorrect logic when producer is removed
jerrypeng
force-pushed
the
fix_exclusive_producer
branch
from
71a7682
to
08a9c96
Compare
jerrypeng
changed the title
| @@ -109,6 +109,9 @@ | |||
|
|
|||
| protected volatile Optional<Long> topicEpoch = Optional.empty(); | |||
| private volatile boolean hasExclusiveProducer; | |||
| // pointer to the exclusive producer | |||
| private volatile Producer exclusiveProducer; | |||
There was a problem hiding this comment.
Should we just use a ref to the name instead? Otherwise it will provide a much longer toString() representation
There was a problem hiding this comment.
We could. Just wanted to preserve the existing logging behavior.
There was a problem hiding this comment.
The existing one was printing the key of the map, hence just the producer name
| return FutureUtil.failedFuture(new ProducerBusyException( | ||
| "Topic has an existing exclusive producer: " + producers.keys().nextElement())); | ||
| return FutureUtil.failedFuture( | ||
| new ProducerFencedException("Topic has an existing exclusive producer: " + exclusiveProducer)); |
There was a problem hiding this comment.
A shared producer will not get fenced, instead it will get a busy error
| @@ -619,17 +629,21 @@ public void removeProducer(Producer producer) { | |||
|
|
|||
| protected void handleProducerRemoved(Producer producer) { | |||
| // decrement usage only if this was a valid producer close | |||
| long newCount = USAGE_COUNT_UPDATER.decrementAndGet(this); | |||
| if (newCount == 0) { | |||
| USAGE_COUNT_UPDATER.decrementAndGet(this); | |||
There was a problem hiding this comment.
Since we add an issue withe consumers here, can you add a test with consumer, to make sure we don’t break it in the future?
* Fix a couple of bugs in exclusive producer 1. Fix uncaught exception in exclusive producer 2. Incorrect logic when producer is removed Co-authored-by: Jerry Peng <jerryp@splunk.com>
Motivation
When multiple producers contend to become the exclusive producer. A race condition can occur and the following exception can happen but gets silently swallowed which eventually leads to a client timeout.
Also
USAGE_COUNT_UPDATERis used incorrectly inhandleProducerRemoved()