[pulsar-broker] Close previous dispatcher when subscription type changes

[massakam]

Motivation

If the subscription type on a topic changes, a new dispatcher is created and the old one is discarded. However, this old dispatcher is not closed. This will cause a memory leak.

Modifications

When the type of a subscription changes and a new dispatcher is created, close the previous one.

[codelipenghui]

I noticed if all consumer disconnected in a subscription, the topic will remove this subscription and the dispatcher is maintained by the subscription. So why don't close previous dispatcher will cause a memory leak, maybe i have not get your point here.

[merlimat]

👍

[merlimat]

nit: for regular get/set operations, we can just use the variable itself return isClosed == TRUE

[massakam]

Fixed.

[massakam]

@codelipenghui

I noticed if all consumer disconnected in a subscription, the topic will remove this subscription and the dispatcher is maintained by the subscription. So why don't close previous dispatcher will cause a memory leak, maybe i have not get your point here.

The problem is that the rate limiter used for dispatch throttling is only closed in the close() method of dispatcher.

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

Lines 374 to 376 in b9e7bf3

	if (dispatchRateLimiter.isPresent()) {
	dispatchRateLimiter.get().close();
	}

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

Lines 334 to 344 in b9e7bf3

	public void close() {
	// close rate-limiter
	if (dispatchRateLimiterOnMessage != null) {
	dispatchRateLimiterOnMessage.close();
	dispatchRateLimiterOnMessage = null;
	}
	if (dispatchRateLimiterOnByte != null) {
	dispatchRateLimiterOnByte.close();
	dispatchRateLimiterOnByte = null;
	}
	}

And if the RateLimiter object is not closed, the renewTask executed every second will continue to run without being cancelled. Therefore, even if the dispatcher is garbage collected, RateLimiter is not removed.

pulsar/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java

Lines 87 to 98 in b9e7bf3

	@Override
	public synchronized void close() {
	if (!isClosed) {
	if (!externalExecutor) {
	executorService.shutdownNow();
	}
	if (renewTask != null) {
	renewTask.cancel(false);
	}
	isClosed = true;
	}
	}

[codelipenghui]

@massakam Got it, thanks. why don't we close the dispatcher at removeConsumer

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

Line 227 in 499069e

if (dispatcher.getConsumers().isEmpty()) {

[massakam]

@codelipenghui Hmm, if the cursor is not durable, the subscription is closed, so the dispatcher should also be closed and I think it should be fixed.

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

Lines 230 to 232 in 499069e

	if (!cursor.isDurable()) {
	// If cursor is not durable, we need to clean up the subscription as well
	close();

I will fix it in another pull-request.

[merlimat]

@massakam Got it, thanks. why don't we close the dispatcher at removeConsumer

when consumers are removed, it might be very likely that they're going to get reconnected immediately thereafter. if the subscription type is not changed, dispatcher will still be valid

[codelipenghui]

retest this please

[massakam]

rerun cpp tests
rerun integration tests