[improve][meta] Fix invalid use of drain API and race condition in closing metadata store

[lhotari]

Motivation

There's currently some memory leaks in tests and while investigating the issue, I found out that there's a large number of uncompleted CompletableFutures in the heap dump.

Currently the metadata store doesn't complete all pending operations when it is closed.
There are multiple problems:

• MpscUnboundedArrayQueue.drain will limit the number of entries to 4096 at a time
• MpscUnboundedArrayQueue.drain uses relaxedPoll which is eventually consistent
• There might be batches in flight, which need to be terminated

Modifications

• replace drain with a while loop in close
• replace drain with a for loop in batch flushing
• add isClosed checks

Additional Context

CompletableFutures in the heap dump:

The instances are related to org.apache.pulsar.broker.resources.NamespaceResources$PartitionedTopicResources$$Lambda$1819+0x00007f08a8b65ee8:

org.apache.pulsar.broker.resources.NamespaceResources$PartitionedTopicResources$$Lambda$1819+0x00007f08a8b65ee8 seems to be this code block:

pulsar/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java

Lines 345 to 374 in bbdc173

	markPartitionedTopicDeletedAsync(topic).whenCompleteAsync((markResult, markExc) -> {
	final boolean mdFound;
	if (markExc != null) {
	if (markExc.getCause() instanceof MetadataStoreException.NotFoundException) {
	mdFound = false;
	} else {
	log.error("Failed to mark the topic {} as deleted", topic, markExc);
	future.completeExceptionally(markExc);
	return;
	}
	} else {
	mdFound = true;
	}
	supplier.get().whenComplete((deleteResult, deleteExc) -> {
	if (deleteExc != null && mdFound) {
	unmarkPartitionedTopicDeletedAsync(topic)
	.thenRun(() -> future.completeExceptionally(deleteExc))
	.exceptionally(ex -> {
	log.warn("Failed to unmark the topic {} as deleted", topic, ex);
	future.completeExceptionally(deleteExc);
	return null;
	});
	} else if (deleteExc != null) {
	future.completeExceptionally(deleteExc);
	} else {
	future.complete(deleteResult);
	}
	});
	});

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[lhotari]

It seems that the OOME is another issue. #22586

[merlimat]

👍

[merlimat]

This one should be ok, since it's already done in a loop: while (!readOps.isEmpty()) {...}