[feat][broker] PIP-451: Support Label-based Topic Subscription

[coderzc]

Fixes #xyz

Main Issue: #xyz

PIP: #xyz

Motivation

Modifications

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

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

Documentation

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

Matching PR in forked repository

PR in forked repository:

[shibd]

I support pushing this feature forward. It effectively simplifies the pattern subscription issue and provides a multi-dimensional approach to topic management.

However, there are some implementation details that require further discussion.

[shibd]

The SystemTopicBasedTopicPoliciesService constructs the cache for a specific namespace based on the assigned namespace bundles.

In production scenarios, the bundles of a single namespace are highly likely to be evenly distributed across all brokers. This implies that every broker will effectively maintain a full copy of the this cache. I am concerned about the potential memory footprint in this scenario

[coderzc]

Yes, each broker needs to cache the full amount of labelTopicInvertedIndex.

[shibd]

What does the data structure of labelMatcher look like? Could you provide an example?

[coderzc]

Like this:

public class LabelMatcher {

    private Map<String, String> labels;

    boolean matches(Map<String, String> targetLabels) {
        for (Map.Entry<String, String> entry : labels.entrySet()) {
            String key = entry.getKey();
            String value = entry.getValue();
            if (!targetLabels.containsKey(key) || !targetLabels.get(key).equals(value)) {
                return false;
            }
        }
        return true;
    }
}

[coderzc]

This can be simplified into a tool function.

[shibd]

If can provide a exmaples for user side, that be better.

[shibd]

This input parameter cannot represent specific labels for a specific namespace. Perhaps we need to wrap the parameter as:

List {
   String namespace,
   Map<String, String> labels
}

[coderzc]

Why is it necessary to use different labels for different namespaces?

[shibd]

Suggested change

	### TopicPoliciesService Interface Changes
	#### TopicPoliciesService Interface Changes

[shibd]

I'm wondering how we should implement this feature, since not every broker maintains a full labelTopicInvertedIndex cache. In the REST API implementation, we would first need to determine which broker is assigned to any bundle of that namespace, and then send the request there.

Is this lookup operation easy to achieve?

[coderzc]

I think the topic policy under a namespace can be loaded onto any broker， since we call prepareInitPoliciesCacheAsync(NamespaceName namespace).