[feat][broker] PIP-469: Legacy-aware topic policies backend routing and metadata-store topic policies - #25707

We might need to consider whether to change the registerListener API because we cannot register listener for both LegacyAwareTopicPoliciesService and SystemTopicBasedTopicPoliciesService.

Let me address these two comments before merging

@dao-jun I've updated the document for why the implementation adopts a different path.

The return value of registerListener (a registration handle) is discarded. close() only clears the local listeners map and invalidates caches, but never unregisters these metadata-store listeners. After close(), the service can still receive and process notifications (the closed flag guards most paths, but the listener
lambdas themselves are still held by the metadata store). Consider capturing and closing the registration handles in close().

Regarding the previous comment, it seems to be wrong. Firstly, registerListener does not return any handle, see

Secondly, it follows the similar pattern of

Though this PR has registered two listeners with a parameter that indicates if the policy is global so that we don't need to check if the notification path starts with GLOBAL_POLICIES_ROOT or LOCAL_POLICIES_ROOT.

Could you double check the review comments if they are generated from LLM?

@nodece Let me merge it first because this topic is out of the scope of the topic policies service itself. I agree that your comments [1] and [2] make sense, but it's more like an issue that should be resolved by structured concurrency, which needs careful design.

It would be better add a common abstraction for listeners and skip all of them after closing the topic policies service.

[1] #25707 (comment)
[2] #25707 (comment)

I tried to write a test to reproduce and then found that even if a listener is registered due to the race, there still might be a real issue.

    private static final CountDownLatch realCloseLatch = new CountDownLatch(1);
    private static final CountDownLatch closeLatch = new CountDownLatch(1);

    public static class MockTopicPoliciesService extends MetadataStoreTopicPoliciesService {

        public void close() {
            super.close();
            realCloseLatch.countDown();
            try {
                if (!closeLatch.await(10, TimeUnit.SECONDS)) {
                    log.info().log("Failed to close TopicPoliciesService within the timeout");
                }
            } catch (InterruptedException e) {
                log.info().log("Interrupted when closing TopicPoliciesService");
            }
        }
    }

The test can be easily written based on the mocked implementation above.

MetadataStoreTopicPoliciesService#close is a very simple operation that changes closed to true. Even if a different broker receives a topic policies update request and updates the policies, handleNotification will skip calling notifyListeners in

Therefore, basically these zombie listeners won't be notified.

In another case, if the CAS of the closed flag happened before listeners were notified, where asynchronous operations had been started, these listeners would be cleared and would not be notified anymore.