You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/02/10 07:42:26 UTC
[GitHub] [pulsar] jerrypeng opened a new pull request #9554: Fix uncaught exception in exclusive producer creation
jerrypeng opened a new pull request #9554:
URL: https://github.com/apache/pulsar/pull/9554
### 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.
21:44:46.038 [ForkJoinPool.commonPool-worker-2] ERROR org.apache.pulsar.broker.service.AbstractTopic - got error: java.util.NoSuchElementException
java.util.NoSuchElementException: null
at java.util.concurrent.ConcurrentHashMap$KeyIterator.next(ConcurrentHashMap.java:3416) ~[?:1.8.0_231]
at java.util.concurrent.ConcurrentHashMap$KeyIterator.nextElement(ConcurrentHashMap.java:3423) ~[?:1.8.0_231]
at org.apache.pulsar.broker.service.AbstractTopic.incrementTopicEpochIfNeeded(AbstractTopic.java:399) ~[pulsar-broker.jar:2.8.0-SNAPSHOT]
at org.apache.pulsar.broker.service.AbstractTopic.addProducer(AbstractTopic.java:346) ~[pulsar-broker.jar:2.8.0-SNAPSHOT]
at org.apache.pulsar.broker.service.persistent.PersistentTopic.addProducer(PersistentTopic.java:509) ~[pulsar-broker.jar:2.8.0-SNAPSHOT]
at org.apache.pulsar.broker.service.ServerCnx.lambda$null$22(ServerCnx.java:1183) ~[pulsar-broker.jar:2.8.0-SNAPSHOT]
at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:656) ~[?:1.8.0_231]
at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:632) ~[?:1.8.0_231]
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_231]
at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_231]
at org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.lambda$null$6(BookkeeperSchemaStorage.java:230) ~[pulsar-broker.jar:2.8.0-SNAPSHOT]
at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:822) [?:1.8.0_231]
at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:797) [?:1.8.0_231]
at java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:443) [?:1.8.0_231]
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) [?:1.8.0_231]
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056) [?:1.8.0_231]
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) [?:1.8.0_231]
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157) [?:1.8.0_231]
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #9554: Fix a couple of bugs in exclusive producer
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #9554:
URL: https://github.com/apache/pulsar/pull/9554#discussion_r574241928
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -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);
Review comment:
sure
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #9554: Fix uncaught exception in exclusive producer creation
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #9554:
URL: https://github.com/apache/pulsar/pull/9554#discussion_r574012085
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -382,17 +382,25 @@ public String getReplicatorPrefix() {
switch (producer.getAccessMode()) {
case Shared:
if (hasExclusiveProducer || !waitingExclusiveProducers.isEmpty()) {
- return FutureUtil.failedFuture(new ProducerBusyException(
- "Topic has an existing exclusive producer: " + producers.keys().nextElement()));
+ StringBuilder errorMsg = new StringBuilder();
+ errorMsg.append("Topic has an existing exclusive producer");
+ if (producers.keys().hasMoreElements()) {
+ errorMsg.append(": " + producers.keys().nextElement());
+ }
Review comment:
```suggestion
producers.keySet().stream().findFirst().map(p -> errorMsg.append(": " ).append(p));
```
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -382,17 +382,25 @@ public String getReplicatorPrefix() {
switch (producer.getAccessMode()) {
case Shared:
if (hasExclusiveProducer || !waitingExclusiveProducers.isEmpty()) {
- return FutureUtil.failedFuture(new ProducerBusyException(
- "Topic has an existing exclusive producer: " + producers.keys().nextElement()));
+ StringBuilder errorMsg = new StringBuilder();
+ errorMsg.append("Topic has an existing exclusive producer");
+ if (producers.keys().hasMoreElements()) {
+ errorMsg.append(": " + producers.keys().nextElement());
+ }
+ return FutureUtil.failedFuture(new ProducerBusyException(errorMsg.toString()));
} else {
// Normal producer getting added, we don't need a new epoch
return CompletableFuture.completedFuture(topicEpoch);
}
case Exclusive:
if (hasExclusiveProducer || !waitingExclusiveProducers.isEmpty()) {
- return FutureUtil.failedFuture(new ProducerFencedException(
- "Topic has an existing exclusive producer: " + producers.keys().nextElement()));
+ StringBuilder errorMsg = new StringBuilder();
+ errorMsg.append("Topic has an existing exclusive producer");
+ if (producers.keys().hasMoreElements()) {
+ errorMsg.append(": " + producers.keys().nextElement());
+ }
Review comment:
```suggestion
producers.keySet().stream().findFirst().map(p -> errorMsg.append(": " ).append(p));
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng merged pull request #9554: Fix a couple of bugs in exclusive producer
Posted by GitBox <gi...@apache.org>.
jerrypeng merged pull request #9554:
URL: https://github.com/apache/pulsar/pull/9554
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #9554: Fix a couple of bugs in exclusive producer
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #9554:
URL: https://github.com/apache/pulsar/pull/9554#discussion_r574253031
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -109,6 +109,9 @@
protected volatile Optional<Long> topicEpoch = Optional.empty();
private volatile boolean hasExclusiveProducer;
+ // pointer to the exclusive producer
+ private volatile Producer exclusiveProducer;
Review comment:
changed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #9554: Fix uncaught exception in exclusive producer creation
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #9554:
URL: https://github.com/apache/pulsar/pull/9554#discussion_r573942728
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -382,17 +382,25 @@ public String getReplicatorPrefix() {
switch (producer.getAccessMode()) {
case Shared:
if (hasExclusiveProducer || !waitingExclusiveProducers.isEmpty()) {
- return FutureUtil.failedFuture(new ProducerBusyException(
- "Topic has an existing exclusive producer: " + producers.keys().nextElement()));
+ StringBuilder errorMsg = new StringBuilder();
+ errorMsg.append("Topic has an existing exclusive producer");
+ if (producers.keys().hasMoreElements()) {
+ errorMsg.append(": " + producers.keys().nextElement());
Review 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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #9554: Fix a couple of bugs in exclusive producer
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #9554:
URL: https://github.com/apache/pulsar/pull/9554#discussion_r574238697
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -109,6 +109,9 @@
protected volatile Optional<Long> topicEpoch = Optional.empty();
private volatile boolean hasExclusiveProducer;
+ // pointer to the exclusive producer
+ private volatile Producer exclusiveProducer;
Review comment:
Should we just use a ref to the name instead? Otherwise it will provide a much longer toString() representation
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -382,17 +385,17 @@ public String getReplicatorPrefix() {
switch (producer.getAccessMode()) {
case Shared:
if (hasExclusiveProducer || !waitingExclusiveProducers.isEmpty()) {
- 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));
Review comment:
A shared producer will not get fenced, instead it will get a busy error
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -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);
Review 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?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #9554: Fix a couple of bugs in exclusive producer
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #9554:
URL: https://github.com/apache/pulsar/pull/9554#discussion_r574250915
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -109,6 +109,9 @@
protected volatile Optional<Long> topicEpoch = Optional.empty();
private volatile boolean hasExclusiveProducer;
+ // pointer to the exclusive producer
+ private volatile Producer exclusiveProducer;
Review comment:
The existing one was printing the key of the map, hence just the producer name
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #9554: Fix a couple of bugs in exclusive producer
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #9554:
URL: https://github.com/apache/pulsar/pull/9554#discussion_r574242046
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -109,6 +109,9 @@
protected volatile Optional<Long> topicEpoch = Optional.empty();
private volatile boolean hasExclusiveProducer;
+ // pointer to the exclusive producer
+ private volatile Producer exclusiveProducer;
Review comment:
We could. Just wanted to preserve the existing logging behavior.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -382,17 +385,17 @@ public String getReplicatorPrefix() {
switch (producer.getAccessMode()) {
case Shared:
if (hasExclusiveProducer || !waitingExclusiveProducers.isEmpty()) {
- 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));
Review comment:
good catch
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org