PIP-145: Improve performance of regex subscriptions

[andrasbeni]

Discussion thread: https://lists.apache.org/thread/51y8s6tfw1p1h5dcb9fgzkzd8tw2bqpz

Motivation

Pulsar allows consumers to subscribe to multiple topics by a pattern. When using this feature, consumers poll brokers for the list of all topics in a namespace and filter the list on the client side based on the pattern. This causes unnecessary network load since most of the time only a small fraction of returned topics match the pattern. In addition polling introduces latency in processing messages produced to a newly created topic.

Goal

This PIP proposes three changes to improve performance and decrease network utilization:

• Server side filtering of topic list: clients will supply the broker with a regex pattern when requesting the list of topics for a namespace. Brokers will apply this pattern when present and only include matching topics in the response.
• Hashing topic list to skip updates: the feature will introduce a hash similar in usage to the HTTP ETag header. The broker computes a hash of the topic list and checks it against the hash the client sends along its request. This allows for short responses that simply mean "Nothing has changed".
• Notifications for faster discovery: clients will be able to register with brokers as observers of the topic list. Brokers will send events to clients whenever there's a change in the list of matching topics.

To help compatibility of new clients with older brokers, a new feature flag will be introduced for this feature. Brokers will return FeatureFlags as part of the CommandConnected message to let clients know what features they support.

First, the feature will be implemented in the broker and the Java client, but later other clients can also make use of the capability.

API Changes

Protocol Changes

New fields will be added to existing commands CommandGetTopicsOfNamespace and CommandGetTopicsOfNamespaceResponse.

message CommandGetTopicsOfNamespace {
    enum Mode {
        PERSISTENT = 0;
        NON_PERSISTENT = 1;
        ALL = 2;
    }
    required uint64 request_id    = 1;
    required string namespace    = 2;
    optional Mode mode = 3 [default = PERSISTENT];
    optional string topics_pattern = 4;
    optional string topics_hash    = 5;
}

message CommandGetTopicsOfNamespaceResponse {
    required uint64 request_id    = 1;
    repeated string topics         = 2;
    // true iff the topic list was filtered by the pattern supplied by the client
    optional bool   filtered       = 3 [default = false];
    // hash computed from the names of matching topics
    optional string topics_hash    = 4;
    // if false, topics is empty and the list of matching topics has not changed
    optional bool   changed        = 5 [default = true];
}

Clients can register as topic list observers by sending the command CommandWatchTopicList:

message CommandWatchTopicList {
    required uint64 watcher_id     = 1;
    required string namespace      = 2;
    required string topics_pattern = 3;
    // Only present when the client reconnects:
    optional string topics_hash    = 4;
}

Brokers will respond with a success message containing the watcher ID and the initial list of topics.

message CommandWatchTopicListSuccess {
    required uint64 watcher_id     = 1;
    repeated string topic          = 2;
    required string topics_hash    = 3;
    // If enableBrokerSideTopicFiltering=false (see later), watchers can be used but filtering is to be done by client
    required boolean filtered  = 4;
}

When new matching topics are added or deleted, the broker sends an update along with the hash computed from the whole list of matching topics (i.e. not just those that are listed in this message).

message CommandWatchTopicUpdate {
    required uint64 watcher_id     = 1;
    repeated string new_topics     = 2;
    repeated string deleted_topics = 3;
    required string topics_hash    = 4;
    // If enableBrokerSideTopicFiltering=false (see later), watchers can be used but filtering is to be done by client
    required boolean filtered  = 5;
}

Clients can unsubscribe the watcher by sending a CommandUnwatchTopicList message, to which the response is a CommandWatchTopicListSuccess without any topics.

message CommandUnwatchTopicList {
    required uint64 watcher_id     = 1;
}

When a client connects to a broker it is notified if the broker supports topic watchers. If not, it will not send CommandWatchTopicList message and continues to rely on polling.

message FeatureFlags {
  optional bool supports_auth_refresh = 1 [default = false];
  optional bool supports_broker_entry_metadata = 2 [default = false];
  optional bool supports_partial_producer = 3 [default = false];
  optional bool supports_topic_watchers = 4 [default = false];
}

message CommandConnected {
    required string server_version = 1;
    optional int32 protocol_version = 2 [default = 0];
    optional int32 max_message_size = 3;
    optional FeatureFlags feature_flags = 4;
}

HTTP clients will not be changed and will continue using the current polling behaviour.

Configuration Changes

A new broker configuration property enableBrokerSideTopicFiltering will be added with default value true. Setting this to false will disable the feature.

Implementation

Polling with pattern

Pulsar clients will poll for topics with the pattern included in the command. Initially the client doesn't have a topicsHash, but once the broker has responded, clients will retain the hash and send it with the next command. If the response from the server contains no hash, the client will perform client side filtering. Otherwise, clients will consider the returned list as already filtered.

The Pulsar broker will check the command for topicsPattern. If there's no pattern in the message, the broker will respond with all topics of the namespace. If a pattern is present, the list of topics is filtered and a hash is computed from the list. If the request contains a topicsHash and it equals the current hash the response will not contain the list of topics, only the changed flag is set to false. The pattern in topicsPattern will be evaluated using java.util.regex.Pattern.

Notifications

If the broker supports topic list watchers, the client will create such a watcher by sending CommandWatchTopicList. A new class, org.apache.pulsar.TopicsListService will keep track of watchers and will listen to changes in the metadata. Whenever a topic is created it checks if any watchers should be notified and sends an update through the ServerCnx. To prevent memory leaks, all watchers will be removed from the TopicsListService when the ServerCnx's channel becomes inactive.

Compatibility

Old clients with new servers

When the server receives a message without the new fields, it will not filter the messages but sends the whole list of topics. These clients will ignore the new fields in the response.

New clients with old servers

New fields in the client request will be ignored by the protobuf parser. The server will send the unfiltered list and omit the new fields (as it does not know about them). The client will check the response for the new fields. If the hash is not present, the client filters the result as it does now.

The introduction of FeatureFlags in CommandConnected will prevent the client from sending CommandWatchTopicList messages to brokers that don't yet support it.

The same is true when broker-side topic filtering is switched off.

Rejected Alternatives

An alternative that came up during the discussion of this PIP is creating a system topic to which topic names are produced by the brokers and consumers subscribe to it to get notified. It was agreed that it would be very hard to make this reliable.

[github-actions]

The issue had no activity for 30 days, mark with Stale label.

[github-actions]

The issue had no activity for 30 days, mark with Stale label.