Hi everyone,
I've just created a proposal for improving the performance of regex
subscriptions.
It's available at https://github.com/apache/pulsar/issues/14505 and is
copied below.
Let me know what you think.
Thanks,
Andras
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 ChangesProtocol 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;
}
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;
}
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.
ImplementationPolling 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.
CompatibilityOld 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.
