[improve][pip] PIP-368: Support lookup based on the lookup properties (…

…#23075)

### Motivation

Currently, the lookup process uses only the topic name as its parameter. However, to enhance this process, it's
beneficial for clients to provide additional information. This could be done by introducing the `lookupProperties` field
in the client configuration. Clients can then share these properties with the broker during lookup.

On the broker side, the broker could also contain some properties that are used for the lookup. We can also support the
lookupProperties for the broker. The broker can use these properties to make a better decision on which broker to
return.

Here is the rack-aware lookup scenario for using the client properties for the lookup:
Assuming there are two brokers that broker-0 configures the lookup property "rack" with "A" and broker-1 configures the
lookup property "rack" with "B". By using the lookup properties, clients can supply rack information during the lookup,
enabling the broker to identify and connect them to the nearest broker within the same rack. If a client that configures
the "rack" property with "A" connects to a lookup broker, the customized load manager can determine broker-0 as the
owner broker since the broker and the client have the same rack property.

### Modifications

Add new configuration `lookupProperties` to the client. While looking up the broker, the client will send the properties
to the broker through `CommandLookupTopic` request.

The `lookupProperties` will then be added to the `LookupOptions`. The Load Manager implementation can access
the `properties` through `LookupOptions` to make a better decision on which broker to return.

The properties are used only when the protocol is the binary protocol, starting with `pulsar://` or `pulsar+ssl://`, or
if the `loadManagerClassName` in the broker is a class that implements the `ExtensibleLoadManager` interface.

To support configuring the `lookupProperties` on the broker side, introduce a new broker
configuration `lookupPropertyPrefix`. Any broker configuration properties that start with the `lookupPropertyPrefix`
will be included into the `BrokerLookupData` and be persisted in the metadata store. The broker can use these properties
during the lookup.

In this way, to support the rack-aware lookup scenario mentioned in the "Motivation" part, the client can set the rack
information in the client `lookupProperties`. Similarly, the broker can also set the rack information in the broker
configuration like `lookup.rack`. The `lookup.rack` will be stored in the `BrokerLookupData`. A customized load manager
can then be implemented. For each lookup request, it will go through the `BrokerLookupData` for all brokers and select
the broker in the same rack to return.

pip/pip-368.md

@@ -0,0 +1,185 @@
+# PIP-368: Support lookup based on the lookup properties
+
+# Background knowledge
+
+## How Pulsar Lookup Works
+
+Before producing or consuming messages, a Pulsar client must first find the broker responsible for the topic. This
+happens through the lookup service. The client sends a `CommandLookupTopic` request with the topic name to the broker
+lookup service.
+
+On the broker side, the broker will register itself to the metadata store using a distributed lock with the value
+of [`BrokerLookupData`](https://github.com/apache/pulsar/blob/7fe92ac43cfd2f2de5576a023498aac8b46c7ac8/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLookupData.java#L34-L44)
+when starting. The lookup service will first choose the owner broker. And then retrieve the `BrokerLookupData` of the
+owner broker and finally return to the client. The client then interacts with this broker to produce or consume
+messages.
+
+Users can customize the lookup process by setting a custom load manager in the `loadManagerClassName` configuration.
+
+# Motivation
+
+Currently, the lookup process uses only the topic name as its parameter. However, to enhance this process, it's
+beneficial for clients to provide additional information. This could be done by introducing the `lookupProperties` field
+in the client configuration. Clients can then share these properties with the broker during lookup.
+
+On the broker side, the broker could also contain some properties that are used for the lookup. We can also support the
+lookupProperties for the broker. The broker can use these properties to make a better decision on which broker to
+return.
+
+Here is the rack-aware lookup scenario for using the client properties for the lookup:
+Assuming there are two brokers that broker-0 configures the lookup property "rack" with "A" and broker-1 configures the
+lookup property "rack" with "B". By using the lookup properties, clients can supply rack information during the lookup,
+enabling the broker to identify and connect them to the nearest broker within the same rack. If a client that configures
+the "rack" property with "A" connects to a lookup broker, the customized load manager can determine broker-0 as the
+owner broker since the broker and the client have the same rack property.
+
+# Goals
+
+## In Scope
+
+- Enable setting up lookup properties in both client and broker configurations.
+- Allow clients to provide extra lookup information to brokers during the lookup process.
+
+## Out of Scope
+
+- The implementation of the rack-aware lookup scenario.
+
+# High Level Design
+
+Add new configuration `lookupProperties` to the client. While looking up the broker, the client will send the properties
+to the broker through `CommandLookupTopic` request.
+
+The `lookupProperties` will then be added to the `LookupOptions`. The Load Manager implementation can access
+the `properties` through `LookupOptions` to make a better decision on which broker to return.
+
+The properties are used only when the protocol is the binary protocol, starting with `pulsar://` or `pulsar+ssl://`, or
+if the `loadManagerClassName` in the broker is a class that implements the `ExtensibleLoadManager` interface.
+
+To support configuring the `lookupProperties` on the broker side, introduce a new broker
+configuration `lookupPropertyPrefix`. Any broker configuration properties that start with the `lookupPropertyPrefix`
+will be included into the `BrokerLookupData` and be persisted in the metadata store. The broker can use these properties
+during the lookup.
+
+In this way, to support the rack-aware lookup scenario mentioned in the "Motivation" part, the client can set the rack
+information in the client `lookupProperties`. Similarly, the broker can also set the rack information in the broker
+configuration like `lookup.rack`. The `lookup.rack` will be stored in the `BrokerLookupData`. A customized load manager
+can then be implemented. For each lookup request, it will go through the `BrokerLookupData` for all brokers and select
+the broker in the same rack to return.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+## Public-facing Changes
+
+### Configuration
+
+Add new configuration `lookupProperties` to the `ClientBuilder`.
+
+```java
+/**
+ * Set the properties used for topic lookup.
+ * <p>
+ * When the broker performs topic lookup, these lookup properties will be taken into consideration in a customized load
+ * manager. 
+ * <p>
+ * Note: The lookup properties are only used in topic lookup when:
+ * - The protocol is binary protocol, i.e. the service URL starts with "pulsar://" or "pulsar+ssl://"
+ * - The `loadManagerClassName` config in broker is a class that implements the `ExtensibleLoadManager` interface
+ */
+ClientBuilder lookupProperties(Map<String, String> properties);
+```
+
+Add new broker configuration `lookupPropertyPrefix` to the `ServiceConfiguration`:
+
+```java
+
+@FieldContext(
+        category = CATEGORY_SERVER,
+        doc = "The properties whose name starts with this prefix will be uploaded to the metadata store for "
+                + " the topic lookup"
+)
+private String lookupPropertyPrefix = "lookup.";
+```
+
+### Binary protocol
+
+Add `properties` field to the `CommandLookupTopic`. Now the `CommandLookupTopic` will look like:
+
+```protobuf
+message KeyValue {
+  required string key = 1;
+  required string value = 2;
+}
+
+message CommandLookupTopic {
+  required string topic = 1;
+  required uint64 request_id = 2;
+  optional bool authoritative = 3 [default = false];
+  optional string original_principal = 4;
+  optional string original_auth_data = 5;
+  optional string original_auth_method = 6;
+  optional string advertised_listener_name = 7;
+  // The properties used for topic lookup
+  repeated KeyValue properties = 8;
+}
+```
+
+When the client lookups a topic, it will set the client `lookupPorperties` to the `CommandLookupTopic.properties`.
+
+### Public API
+
+Currently, there is a public method `assign` in the `ExtensibleLoadManager` interface that will accept
+the `LookupOptions` to lookup the topic.
+
+```java
+public interface ExtensibleLoadManager {
+    CompletableFuture<Optional<BrokerLookupData>> assign(Optional<ServiceUnitId> topic,
+                                                         ServiceUnitId serviceUnit,
+                                                         LookupOptions options);
+}
+```
+
+In this proposal, the `properties` will be added to the `LookupOptions`:
+
+```java
+public class LookupOptions {
+    // Other fields are omitted ...
+
+    // The properties used for topic lookup
+    private final Map<String, String> properties;
+}
+```
+
+The `LookupOptions.properties` will be set to the value of `CommandLookupTopic.properties`.
+This way, the custom `ExtensibleLoadManager` implementation can retrieve the `properties` from the `LookupOptions` to
+make a better decision on which broker to return.
+
+# Monitoring
+
+No new metrics are added in this proposal.
+
+# Security Considerations
+
+No new security considerations are added in this proposal.
+
+# Backward & Forward Compatibility
+
+## Revert
+
+No changes are needed to revert to the previous version.
+
+## Upgrade
+
+No other changes are needed to upgrade to the new version.
+
+# Alternatives
+
+None
+
+# General Notes
+
+# Links
+
+* Mailing List discussion thread: https://lists.apache.org/thread/7n2gncxk3c5q8dxj8fw9y5gcwg6jjg6z
+* Mailing List voting thread: https://lists.apache.org/thread/z0t3dyqj27ldm8rs6nl5jon152ohghvw