In LockManager use concurrent hash map to handle locks notifiications

[merlimat]

Motivation

In LockManagerImpl, the notification will use the locks set to re-check the status of the lock against the metadata store. The access needs to be synchronized because new locks could be added again to the list from other threads.

That leads to these exceptions:

08:50:22.921 [metadata-store-5-1] ERROR org.apache.pulsar.metadata.impl.AbstractMetadataStore - Failed to process metadata store notification
java.util.ConcurrentModificationException: null
	at java.util.HashMap$KeySpliterator.forEachRemaining(HashMap.java:1610) ~[?:?]
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484) ~[?:?]
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474) ~[?:?]
	at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150) ~[?:?]
	at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173) ~[?:?]
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
	at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497) ~[?:?]
	at org.apache.pulsar.metadata.coordination.impl.LockManagerImpl.handleDataNotification(LockManagerImpl.java:119) ~[classes/:?]
	at org.apache.pulsar.metadata.impl.AbstractMetadataStore.lambda$receivedNotification$0(AbstractMetadataStore.java:149) ~[classes/:?]
	at java.util.concurrent.CopyOnWriteArrayList.forEach(CopyOnWriteArrayList.java:804) ~[?:?]
	at org.apache.pulsar.metadata.impl.AbstractMetadataStore.lambda$receivedNotification$1(AbstractMetadataStore.java:147) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1771) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-common-4.1.63.Final.jar:4.1.63.Final]
	at java.lang.Thread.run(Thread.java:830) [?:?]

Additionally, we should use a Map instead of a Set to avoid going through the list when one single node is deleted.

[eolivelli]

LGTM