[fix][test] Fix CoordinatorService, MetadataStore and MockZooKeeper leaks in tests

[nicoloboschi]

Motivation

There are some tests that occasionally fails with non-sense mockito errors like this one:

org.apache.pulsar.broker.PulsarServerException: 
org.mockito.exceptions.misusing.WrongTypeOfReturnValue: 
BrokerService cannot be returned by getConfiguration()
getConfiguration() should return ServiceConfiguration
***
If you're unsure why you're getting above error read on.
Due to the nature of the syntax above problem might occur because:
1. This exception *might* occur in wrongly written multi-threaded tests.
   Please refer to Mockito FAQ on limitations of concurrency testing.
2. A spy is stubbed using when(spy.foo()).then() syntax. It is safer to stub spies - 
   - with doReturn|Throw() family of methods. More in javadocs for Mockito.spy() method.

	at org.apache.pulsar.broker.service.BrokerService.close(BrokerService.java:685)
	at java.base/java.lang.invoke.MethodHandle.invokeWithArguments(MethodHandle.java:710)
	at org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
Caused by: org.mockito.exceptions.misusing.WrongTypeOfReturnValue: 
BrokerService cannot be returned by getConfiguration()
getConfiguration() should return ServiceConfiguration
***
If you're unsure why you're getting above error read on.
Due to the nature of the syntax above problem might occur because:
1. This exception *might* occur in wrongly written multi-threaded tests.
   Please refer to Mockito FAQ on limitations of concurrency testing.
2. A spy is stubbed using when(spy.foo()).then() syntax. It is safer to stub spies - 
   - with doReturn|Throw() family of methods. More in javadocs for Mockito.spy() method.

	at org.apache.pulsar.broker.PulsarService.getConfiguration(PulsarService.java:608)
	at org.apache.pulsar.broker.service.BrokerService.shutdownEventLoopGracefully(BrokerService.java:846)
	at java.base/java.lang.invoke.MethodHandle.invokeWithArguments(MethodHandle.java:710)
	at org.apache.pulsar.broker.service.BrokerService.shutdownEventLoopGracefully(BrokerService.java:846)
	at org.apache.pulsar.broker.service.BrokerService.closeAsync(BrokerService.java:757)
	at java.base/java.lang.invoke.MethodHandle.invokeWithArguments(MethodHandle.java:710)
	at org.apache.pulsar.broker.service.BrokerService.closeAsync(BrokerService.java:720)
	at org.apache.pulsar.broker.service.BrokerService.close(BrokerService.java:680)
	... 2 more

After investigating I noticed that MetadataStore threads can still receive notifications even if the current test method is finished. Also other thread pools are not shutdown in the correct way and the test must wait for their terminations in order to avoid mixing up Mockito mocked objects.

Modifications

• MockZookKeeper close method now really shutdown ZK
• MetadataStore is always closed in MockedBookKeeperTestCase based tests
• Fixed a couple of tests which manually create executors and Netty event loop groups and reordered the resources disposal (the metadatastore must be closed after all)

This is related to past flaky-tests like:

• [branch-2.9] Flaky-test: PersistentTopicStreamingDispatcherTest #13808
• Add null check to workaround NPE in unit tests with Mockito/PowerMock #14006
• Flaky-test: Spy creation sporadically fails in some tests and causes Mockito misuse warning #13620

fix #15774

• no-need-doc

[Technoboy-]

Will reopen later to re-run the test.

[nicoloboschi]

/pulsarbot rerun-failure-checks

[nicoloboschi]

Test failed is not related, I opened an issue: #15676

[lhotari]

Nice improvement!

[nicoloboschi]

/pulsarbot rerun-failure-checks

[nicoloboschi]

/pulsarbot rerun-failure-checks

[lhotari]

@nicoloboschi Please rebase this PR.

It looks like this could help with these issues which happen in PulsarFunctionTlsTest.tearDown:

"main" #1 prio=5 os_prio=0 cpu=67703.73ms elapsed=3524.96s tid=0x00007f0b9c0243f0 nid=0xafb waiting on condition  [0x00007f0ba017c000]
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park(java.base@17.0.3/Native Method)
	- parking to wait for  <0x00000000ce100010> (a java.util.concurrent.CountDownLatch$Sync)
	at java.util.concurrent.locks.LockSupport.park(java.base@17.0.3/LockSupport.java:211)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(java.base@17.0.3/AbstractQueuedSynchronizer.java:715)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(java.base@17.0.3/AbstractQueuedSynchronizer.java:1047)
	at java.util.concurrent.CountDownLatch.await(java.base@17.0.3/CountDownLatch.java:230)
	at org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl.shutdown(ManagedLedgerFactoryImpl.java:635)
	at org.apache.pulsar.broker.ManagedLedgerClientFactory.close(ManagedLedgerClientFactory.java:134)
	at org.apache.pulsar.broker.PulsarService.closeAsync(PulsarService.java:460)
	at org.apache.pulsar.broker.PulsarService.close(PulsarService.java:374)
	at org.apache.pulsar.functions.worker.PulsarFunctionTlsTest.tearDown(PulsarFunctionTlsTest.java:182)

This is the 4th most flaky issue currently.

[github-actions]

The pr had no activity for 30 days, mark with Stale label.

[nicoloboschi]

@lhotari PTAL