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