Fix: deadlock while closing non-shared consumer

[rdhabalia]

Motivation

Fix possible deadlock while closing non-shared consumer.

main" #1 prio=5 os_prio=31 tid=0x00007fad6e802800 nid=0x1c03 waiting on condition [0x000070000c802000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000007416a9818> (a org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section)
        at java.util.concurrent.locks.StampedLock.acquireWrite(StampedLock.java:1119)
        at java.util.concurrent.locks.StampedLock.writeLock(StampedLock.java:354)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.remove(ConcurrentOpenHashMap.java:305)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.access$200(ConcurrentOpenHashMap.java:181)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.remove(ConcurrentOpenHashMap.java:136)
        at org.apache.pulsar.broker.service.persistent.PersistentTopic.removeSubscription(PersistentTopic.java:650)
        at org.apache.pulsar.broker.service.persistent.PersistentSubscription.removeConsumer(PersistentSubscription.java:156)
        - locked <0x00000007452139d8> (a org.apache.pulsar.broker.service.persistent.PersistentSubscription)
        at org.apache.pulsar.broker.service.Consumer.close(Consumer.java:358)
        at org.apache.pulsar.broker.service.Consumer.disconnect(Consumer.java:366)
        at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer$$Lambda$471/975941670.accept(Unknown Source)
        at java.util.concurrent.CopyOnWriteArrayList.forEach(CopyOnWriteArrayList.java:890)
        at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.disconnectAllConsumers(AbstractDispatcherSingleActiveConsumer.java:189)
        - locked <0x0000000745213aa8> (a org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer)
        at org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer.close(AbstractDispatcherSingleActiveConsumer.java:176)
        at org.apache.pulsar.broker.service.persistent.PersistentSubscription.disconnect(PersistentSubscription.java:535)
        - locked <0x00000007452139d8> (a org.apache.pulsar.broker.service.persistent.PersistentSubscription)
        at org.apache.pulsar.broker.service.persistent.PersistentTopic.lambda$16(PersistentTopic.java:760)
        at org.apache.pulsar.broker.service.persistent.PersistentTopic$$Lambda$470/114109618.accept(Unknown Source)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:386)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:160)
        at org.apache.pulsar.broker.service.persistent.PersistentTopic.close(PersistentTopic.java:760)
        at org.apache.pulsar.broker.service.BrokerService$$Lambda$466/399685190.apply(Unknown Source)
        at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:981)
        at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2124)
        at org.apache.pulsar.broker.service.BrokerService.lambda$22(BrokerService.java:924)
        at org.apache.pulsar.broker.service.BrokerService$$Lambda$465/1261714285.accept(Unknown Source)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:386)
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:160)
        at org.apache.pulsar.broker.service.BrokerService.unloadServiceUnit(BrokerService.java:919)
        at org.apache.pulsar.broker.namespace.OwnedBundle.handleUnloadRequest(OwnedBundle.java:125)
        at org.apache.pulsar.broker.namespace.NamespaceService.unloadNamespaceBundle(NamespaceService.java:487)
        at org.apache.pulsar.broker.service.BrokerService.lambda$3(BrokerService.java:408)
        at org.apache.pulsar.broker.service.BrokerService$$Lambda$464/659566172.accept(Unknown Source)
        at java.lang.Iterable.forEach(Iterable.java:75)
        at org.apache.pulsar.broker.service.BrokerService.unloadNamespaceBundlesGracefully(BrokerService.java:405)
        at org.apache.pulsar.broker.service.BrokerService.close(BrokerService.java:360)
        at org.apache.pulsar.broker.PulsarService.close(PulsarService.java:211)

[merlimat]

👍