Fix partitioned system topic check bug

[hangc0276]

Motivation

When checking a partitioned topic whether a system topic, it will always be false. The check logic is.

static boolean isSystemTopic(TopicName topicName) {
        return EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME.equals(topicName.getLocalName());
    }

public static final String NAMESPACE_EVENTS_LOCAL_NAME = "__change_events";

The partitioned topic name is like __change_events-partition-0.

Modification

1. Trim the partition suffix for the topic local name if exists.
2. Add a test to cover this case.

[linlinnn]

what about
topicName.getLocalName().startsWith(EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME);

[315157973]

+1

return topicName != null && topicName.getLocalName().startsWith(EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME);

[315157973]

or

return EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME.equals(topicName.getPartitionedTopicName());

[hangc0276]

what about
topicName.getLocalName().startsWith(EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME);

@linlinnn if the topic name is __change_events_v1, using startsWith, it will return true. However it isn't system topic.

[hangc0276]

or

return EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME.equals(topicName.getPartitionedTopicName());

@315157973 topicName.getPartitionedTopicName() method will return topic name with prefix, such as, persistent://public/defualt/__change_events, it will always return false.

[315157973]

TopicName.get(topicName.getPartitionedTopicName()).getLocalName()

[codelipenghui]

We have a method in the TopicName is isPartitioned(). You can check as followings

if (topicName.isPartitioned()) {
    return EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME.equals(TopicName.get(topicName.getPartitionedTopicName));
} else {
    return 
 EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME.equals(topicName.getLocalName());
}

[hangc0276]

@315157973 good job. I have update the code, PTAL.

[hangc0276]

@codelipenghui I have update the code, PTAL.

[315157973]

LGTM

[dockerzhang]

+1