[pulsar-broker] If cursor is not durable, close dispatcher when all consumers are removed from subscription

[massakam]

Motivation

If cursor is not durable, the subscription is closed and removed from the topic when all consumers are removed.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

Lines 237 to 254 in 413ba03

	public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
	cursor.updateLastActive();
	if (dispatcher != null) {
	dispatcher.removeConsumer(consumer);
	}
	if (dispatcher.getConsumers().isEmpty()) {
	deactivateCursor();
	if (!cursor.isDurable()) {
	// If cursor is not durable, we need to clean up the subscription as well
	close();
	// when topic closes: it iterates through concurrent-subscription map to close each subscription. so,
	// topic.remove again try to access same map which creates deadlock. so, execute it in different thread.
	topic.getBrokerService().pulsar().getExecutor().submit(() ->{
	topic.removeSubscription(subName);
	});
	}
	}

As mentioned in #5288, the dispatcher also needs to be closed at this time. Otherwise, RateLimiter instances will not be garbage collected, causing a memory leak.

[massakam]

rerun java8 tests
rerun integration tests