[fix] [broker] Fix break change: could not subscribe partitioned topic with a suffix-matched regexp due to a mistake of PIP-145

[poorbarcode]

Motivation

• Before PIP-145, it used the Partitioned topic name to match the Regexp. For example, the partitioned topic public/default/tp has one partition named public/default/tp-partition-0. Pulsar will use public/default/tp to match the Regexp^[1].
• After PIP-145, it used the partition name to match the Regexp. For example, the partitioned topic public/default/tp has one partition named public/default/tp-partition-0. Pulsar will use public/default/tp-partition-0 to match the Regexp^[2].

It is a break change that is not expected; these changes are due to a mistake, and we should correct it.

Issue

If a consumer tries to subscribe to a partitioned topic with a suffix-matched regexp, it does not work.

• create topic persistent://public/default/tp
• create a consumer with regexp persistent://public/default/tp$
• the pattern consumer has no internal consumers.

You can reproduce the issue by the test testPreciseRegexpSubscribe.

• it can work when using 2.10.x
• it can not work when using >=2.11.0

Modifications

Revert the behavior as the original implementation(before PIP-145)

Footnotes

[1]: the implementation of 2.10.x

• Pulsar did the check of regexp on the client-side.
• BinaryProtoLookupService.getTopicsUnderNamespace merge partitions to one partitioned topic. E.g.) Pulsar client merges public/default/tp-partition-0 and public/default/tp-partition-1 to ``public/default/tp`
• PulsarClientImpl.topicsPatternFilter Use the partitioned topic name to match the regexp.

BinaryProtoLookupService.getTopicsUnderNamespace: https://github.com/apache/pulsar/blob/branch-2.10/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java#L332-L338

List<String> result = new ArrayList<>();
r.forEach(topic -> {
    String filtered = TopicName.get(topic).getPartitionedTopicName();
    if (!result.contains(filtered)) {
        result.add(filtered);
    }
});

PulsarClientImpl.topicsPatternFilter : https://github.com/apache/pulsar/blob/branch-2.10/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java#L568-L577

public static List<String> topicsPatternFilter(List<String> original, Pattern topicsPattern) {
    final Pattern shortenedTopicsPattern = topicsPattern.toString().contains("://")
        ? Pattern.compile(topicsPattern.toString().split("\\:\\/\\/")[1]) : topicsPattern;

    return original.stream()
        .map(TopicName::get)
        .map(TopicName::toString)
        .filter(topic -> shortenedTopicsPattern.matcher(topic.split("\\:\\/\\/")[1]).matches())
        .collect(Collectors.toList());
}

[2]: the implementation of >=2.11.0

• Pulsar did the check of regexp on the broker-side.
• TopicList.filterTopics() uses the partition name to match the regexp directly. See: https://github.com/apache/pulsar/blob/master/pulsar-common/src/main/java/org/apache/pulsar/common/topics/TopicList.java#L53-L57

public static List<String> filterTopics(List<String> original, Pattern topicsPattern) {

    final Pattern shortenedTopicsPattern = topicsPattern.toString().contains(SCHEME_SEPARATOR)
            ? Pattern.compile(SCHEME_SEPARATOR_PATTERN.split(topicsPattern.toString())[1]) : topicsPattern;

    return original.stream()
            .map(TopicName::get)
            .map(TopicName::toString)
            .filter(topic -> shortenedTopicsPattern.matcher(SCHEME_SEPARATOR_PATTERN.split(topic)[1]).matches())
            .collect(Collectors.toList());
}

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[codelipenghui]

/pulsarbot run-failure-checks

[codecov-commenter]

Codecov Report

Attention: 2 lines in your changes are missing coverage. Please review.

Comparison is base (176bdea) 73.58% compared to head (a0eeae9) 73.52%.
Report is 5 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #21885      +/-   ##
============================================
- Coverage     73.58%   73.52%   -0.06%     
- Complexity    32325    32345      +20     
============================================
  Files          1859     1859              
  Lines        138263   138359      +96     
  Branches      15153    15159       +6     
============================================
- Hits         101736   101732       -4     
- Misses        28644    28743      +99     
- Partials       7883     7884       +1     

Flag	Coverage Δ
inttests	24.12% <0.00%> (+0.02%)	⬆️
systests	23.67% <63.63%> (-0.11%)	⬇️
unittests	72.82% <81.81%> (-0.03%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...ava/org/apache/pulsar/common/topics/TopicList.java	89.65% <81.81%> (-5.59%)	⬇️

... and 62 files with indirect coverage changes

[lhotari]

This breaks Pulsar SQL / Trino tests in branch-3.1 and before. I created #21976 to address the problem.