pip/pip-368.md

# 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 `lookupProperties` 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