[fix][broker] Fix PulsarService.closeAsync where Condition.signalAll was called without holding a lock

[zhaizhibo]

Motivation

In PulsarService.closeAsync(), the closeFuture.handle() callback calls isClosedCondition.signalAll() without acquiring the associated mutex lock.

Modifications

• Fix: Wrap isClosedCondition.signalAll() inside mutex.lock()/unlock() in the closeFuture.handle() callback.
• Test: Add testWaitUntilClosedConcurrentWithCloseAsync in PulsarServiceCloseTest that calls waitUntilClosed() concurrently with closeAsync().

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[zhaizhibo]

@merlimat Fixed.

[lhotari]

LGTM, good catch! This explains some flakiness that has been related to PulsarService.closeAsync in tests.