[Broker] Fix race conditions in closing producers and consumers
closing ServerCnx while producers or consumers are created can lead
to a producer or consumer never getting removed from the topic's
list of producers
@@ -292,6 +292,11 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
// Connection is gone, close the producers immediately
producers.forEach((__, producerFuture) -> {
+ // prevent race conditions in completing producers
+ if (!producerFuture.isDone()
+ && producerFuture.completeExceptionally(new IllegalStateException("Connection closed."))) {
+ return;
+ }
if (producerFuture.isDone() && !producerFuture.isCompletedExceptionally()) {
Producer producer = producerFuture.getNow(null);
producer.closeNow(true);
@@ -299,18 +304,19 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
});
consumers.forEach((__, consumerFuture) -> {
- Consumer consumer;
- if (consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
- consumer = consumerFuture.getNow(null);
- } else {
+ // prevent race conditions in completing consumers
+ if (!consumerFuture.isDone()
+ && consumerFuture.completeExceptionally(new IllegalStateException("Connection closed."))) {
return;
}
-
+ if (consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
+ Consumer consumer = consumerFuture.getNow(null);
try {
consumer.close();
} catch (BrokerServiceException e) {
log.warn("Consumer {} was already closed: {}", consumer, e);
}
+ }
});
this.service.getPulsarStats().recordConnectionClose();
}
