pip/pip-344.md - pulsar - Git at Google

 # PIP-344: Correct the behavior of the public API pulsarClient.getPartitionsForTopic(topicName)

 # Background knowledge

 **Topic auto-creation**
 - The partitioned topic auto-creation is dependent on `pulsarClient.getPartitionsForTopic`
     - It triggers partitioned metadata creation by `pulsarClient.getPartitionsForTopic`
     - And triggers the topic partition creation by producers' registration and consumers' registration.
 - When calling `pulsarClient.getPartitionsForTopic(topicName)`, Pulsar will automatically create the partitioned topic metadata if it does not exist, either using `HttpLookupService` or `BinaryProtoLookupService`.

 **Now `pulsarClient.getPartitionsForTopic`'s behavior**
 | case | broker allow `auto-create` | param allow <br> `create if not exists` | non-partitioned topic | partitioned topic |  current behavior |
 | --- | --- | --- | --- | --- | --- |
 | 1 | `true/false` | `true/false` | `exists: true` | | REST API: `partitions: 0`<br> Binary API: `partitions: 0` |
 | 2 | `true/false` | `true/false` | | `exists: true` <br> `partitions: 3` | REST API: `partitions: 3`<br> Binary API: `partitions: 3` |
 | 3 | `true` | `true` | | | REST API: <br> &nbsp;&nbsp;- `create new: true` <br> &nbsp;&nbsp;- `partitions: 3` <br> Binary API: <br> &nbsp;&nbsp;- `create new: true` <br> &nbsp;&nbsp;- `partitions: 3` <br> |
 | 4 | `true` | `false` | | | REST API: <br> &nbsp;&nbsp;- `create new: false` <br> &nbsp;&nbsp;- `partitions: 0` <br> Binary API: <br> &nbsp;&nbsp;not support <br> |
 | 5 | `false` | `true` | | | REST API: <br> &nbsp;&nbsp;- `create new: false` <br> &nbsp;&nbsp;- `partitions: 0` <br> Binary API: <br> &nbsp;&nbsp;- `create new: false` <br> &nbsp;&nbsp;- `partitions: 0` <br> |

 - Broker allows `auto-create`: see also the config `allowAutoTopicCreation` in `broker.conf`.
 - Param allow <br> `create if not exists`
   - Regarding the HTTP API `PersistentTopics.getPartitionedMetadata`, it is an optional param which named `checkAllowAutoCreation,` and the default value is `false`.
   - Regarding the `pulsar-admin` API, it depends on the HTTP API `PersistentTopics.getPartitionedMetadata`, and it always sets the param `checkAllowAutoCreation` to `false` and can not be set manually.
   - Regarding the client API `HttpLookupService.getPartitionedTopicMetadata`, it depends on the HTTP API `PersistentTopics.getPartitionedMetadata`, and it always sets the param `checkAllowAutoCreation` to `true` and can not be set manually.
   - Regarding the client API `BinaryProtoLookupService.getPartitionedTopicMetadata`, it always tries to create partitioned metadata.
 - `REST API & HTTP API`: Since there are only two implementations of the 4 ways to get partitioned metadata, we call HTTP API `PersistentTopics.getPartitionedMetadata`, `pulsar-admin`, and `HttpLookupService.getPartitionedTopicMetadata` HTTP API, and call `BinaryProtoLookupService.getPartitionedTopicMetadata` Binary API.

 # Motivation

 The param `create if not exists` of the Binary API is always `true.`

 - For case 4 of `pulsarClient.getPartitionsForTopic`'s behavior, it always tries to create the partitioned metadata, but the API name is `getxxx`.
 - For case 5 of `pulsarClient.getPartitionsForTopic`'s behavior, it returns a `0` partitioned metadata, but the topic does not exist. For the correct behavior of this case, we had discussed [here](https://github.com/apache/pulsar/issues/8813) before.
 - BTW, [flink-connector-pulsar](https://github.com/apache/flink-connector-pulsar/blob/main/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java#L221-L227) is using this API to create partitioned topic metadata.

 # Goals

 - Regarding the case 4: Add a new API `PulsarClient.getPartitionsForTopic(String, boolean)` to support the feature that just get partitioned topic metadata and do not try to create one. See detail below.
 - Regarding the case 5: Instead of returning a `0` partitioned metadata, respond to a not found error when calling `pulsarClient.getPartitionsForTopic(String)` if the topic does not exist.

 # Detailed Design

 ## Public-facing Changes

 When you call the public API `pulsarClient.getPartitionsForTopic`, pulsar will not create the partitioned metadata anymore.

 ### Public API
 **LookupService.java**
 ```java

 - CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(TopicName topicName);

 /**
  * 1. Get the partitions if the topic exists. Return "{partition: n}" if a partitioned topic exists; return "{partition: 0}" if a non-partitioned topic exists.
  * 2. When {@param createIfAutoCreationEnabled} is "false," neither partitioned topic nor non-partitioned topic does not exist. You will get an {@link PulsarClientException.NotFoundException}.
  *  2-1. You will get a {@link PulsarClientException.NotSupportedException} if the broker's version is an older one that does not support this feature and the Pulsar client is using a binary protocol "serviceUrl".
  * 3. When {@param createIfAutoCreationEnabled} is "true," it will trigger an auto-creation for this topic(using the default topic auto-creation strategy you set for the broker), and the corresponding result is returned. For the result, see case 1.
  * @version 3.3.0
  */
 + CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(TopicName topicName, boolean createIfAutoCreationEnabled);
 ```

 The behavior of the new API `LookupService.getPartitionedTopicMetadata(TopicName, boolean)`.

 | case | client-side param: `createIfAutoCreationEnabled` | non-partitioned topic | partitioned topic                   | broker-side: topic auto-creation strategy                                                                          | current behavior                                                                            |
 |------|--------------------------------------------------|-----------------------|-------------------------------------|--------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------|
 | 1    | `true/false`                                     | `exists: true`        |                                     |                                                                                                                    | REST/Binary API: `{partitions: 0}`                                                          |
 | 2    | `true/false`                                     |                       | `exists: true` <br> `partitions: 2` |                                                                                                                    | REST/Binary API: `{partitions: 2}`                                                          |
 | 3    | `true`                                           |                       |                                     | `allowAutoTopicCreation`: `true` <be> `allowAutoTopicCreationType`: `non-partitioned`                              | REST/Binary API: <br> &nbsp;&nbsp;- `create new: true` <br> &nbsp;&nbsp;- `{partitions: 0}` |
 | 4    | `true`                                           |                       |                                     | `allowAutoTopicCreation`: `true` <be> `allowAutoTopicCreationType`: `partitioned` <br> `defaultNumPartitions`: `2` | REST/Binary API: <br> &nbsp;&nbsp;- `create new: true` <br> &nbsp;&nbsp;- `{partitions: 2}` |
 | 5    | `false`                                          |                       |                                     | `allowAutoTopicCreation`: `true`                                                                                   | REST/Binary API: <br> &nbsp;&nbsp;- Not found error                                         |
 | 6    | `true`                                           |                       |                                     | `allowAutoTopicCreation`: `false`                                                                                  | REST/Binary API: <br> &nbsp;&nbsp;- Not found error                                         |


 **PulsarClient.java**
 ```java
 // This API existed before. Not change it, thus ensuring compatibility.
 + @Deprecated it is not suggested to use now; please use {@link #getPartitionsForTopic(TopicName, boolean)}.
 - CompletableFuture<List<String>> getPartitionsForTopic(String topic);
 + default CompletableFuture<List<String>> getPartitionsForTopic(String topic) {
 +    getPartitionsForTopic(topic, true);
 + }

 /**
  * 1. Get the partitions if the topic exists. Return "[{partition-0}, {partition-1}....{partition-n}}]" if a partitioned topic exists; return "[{topic}]" if a non-partitioned topic exists.
  * 2. When {@param createIfAutoCreationEnabled} is "false", neither the partitioned topic nor non-partitioned topic does not exist. You will get an {@link PulsarClientException.NotFoundException}.
  *  2-1. You will get a {@link PulsarClientException.NotSupportedException} if the broker's version is an older one that does not support this feature and the Pulsar client is using a binary protocol "serviceUrl".
  * 3. When {@param createIfAutoCreationEnabled} is "true," it will trigger an auto-creation for this topic(using the default topic auto-creation strategy you set for the broker), and the corresponding result is returned. For the result, see case 1.
  * @version 3.3.0
  */
 CompletableFuture<List<String>> getPartitionsForTopic(String topic, boolean createIfAutoCreationEnabled);
 ```

 The behavior of the new API `PulsarClient.getPartitionsForTopic(String, boolean)`.

 | case | client-side param: `createIfAutoCreationEnabled` | non-partitioned topic | partitioned topic                   | broker-side: topic autp-creation strategy                                                                          | current behavior                                                                                                                                |
 |------|--------------------------------------------------|----------------------|-------------------------------------|--------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------|
 | 1    | `true/false`                                     | `exists: true`       |                                     |                                                                                                                    | REST/Binary API: `["{tenat}/{ns}/topic"]`                                                                                                       |
 | 2    | `true/false`                                     |                      | `exists: true` <br> `partitions: 2` |                                                                                                                    | REST/Binary `API`: `["{tenat}/{ns}/topic-partition-0", "{tenat}/{ns}/topic-partition-1"]`                                                       |
 | 3    | `true`                                           |                      |                                     | `allowAutoTopicCreation`: `true` <be> `allowAutoTopicCreationType`: `non-partitioned`                              | REST/Binary API: <br> &nbsp;&nbsp;- `create new: true` <br> &nbsp;&nbsp;- `["{tenat}/{ns}/topic"]`                                              |
 | 4    | `true`                                           |                      |                                     | `allowAutoTopicCreation`: `true` <be> `allowAutoTopicCreationType`: `partitioned` <br> `defaultNumPartitions`: `2` | REST/Binary API: <br> &nbsp;&nbsp;- `create new: true` <br> &nbsp;&nbsp;- `["{tenat}/{ns}/topic-partition-0", "{tenat}/{ns}/topic-partition-1"]` |
 | 5    | `false`                                          |                      |                                     | `allowAutoTopicCreation`: `true`                                                                                   | REST/Binary API: <br> &nbsp;&nbsp;- Not found error                                                                                             |
 | 5    | `true`                                           |                      |                                     | `allowAutoTopicCreation`: `false`                                                                                  | REST/Binary API: <br> &nbsp;&nbsp;- Not found error                                                                                             |


 ### Binary protocol

 **CommandPartitionedTopicMetadata**
 ```
 message CommandPartitionedTopicMetadata {
   + optional bool metadata_auto_creation_enabled = 6 [default = true];
 }
 ```

 **FeatureFlags**
 ```
 message FeatureFlags {
   + optional bool supports_binary_api_get_partitioned_meta_with_param_created_false = 5 [default = false];
 }
 ```

 # Backward & Forward Compatibility

 - Old version client and New version Broker: The client will call the old API.

 - New version client and Old version Broker: The feature flag `supports_binary_api_get_partitioned_meta_with_param_created_false` will be `false`. The client will get a not-support error if the param `createIfAutoCreationEnabled` is false.
	# PIP-344: Correct the behavior of the public API pulsarClient.getPartitionsForTopic(topicName)

	# Background knowledge

	Topic auto-creation
	- The partitioned topic auto-creation is dependent on `pulsarClient.getPartitionsForTopic`
	- It triggers partitioned metadata creation by `pulsarClient.getPartitionsForTopic`
	- And triggers the topic partition creation by producers' registration and consumers' registration.
	- When calling `pulsarClient.getPartitionsForTopic(topicName)`, Pulsar will automatically create the partitioned topic metadata if it does not exist, either using `HttpLookupService` or `BinaryProtoLookupService`.

	Now `pulsarClient.getPartitionsForTopic`'s behavior
	\| case \| broker allow `auto-create` \| param allow <br> `create if not exists` \| non-partitioned topic \| partitioned topic \| current behavior \|
	\| --- \| --- \| --- \| --- \| --- \| --- \|
	\| 1 \| `true/false` \| `true/false` \| `exists: true` \| \| REST API: `partitions: 0`<br> Binary API: `partitions: 0` \|
	\| 2 \| `true/false` \| `true/false` \| \| `exists: true` <br> `partitions: 3` \| REST API: `partitions: 3`<br> Binary API: `partitions: 3` \|
	\| 3 \| `true` \| `true` \| \| \| REST API: <br>   - `create new: true` <br>   - `partitions: 3` <br> Binary API: <br>   - `create new: true` <br>   - `partitions: 3` <br> \|
	\| 4 \| `true` \| `false` \| \| \| REST API: <br>   - `create new: false` <br>   - `partitions: 0` <br> Binary API: <br>   not support <br> \|
	\| 5 \| `false` \| `true` \| \| \| REST API: <br>   - `create new: false` <br>   - `partitions: 0` <br> Binary API: <br>   - `create new: false` <br>   - `partitions: 0` <br> \|

	- Broker allows `auto-create`: see also the config `allowAutoTopicCreation` in `broker.conf`.
	- Param allow <br> `create if not exists`
	- Regarding the HTTP API `PersistentTopics.getPartitionedMetadata`, it is an optional param which named `checkAllowAutoCreation,` and the default value is `false`.
	- Regarding the `pulsar-admin` API, it depends on the HTTP API `PersistentTopics.getPartitionedMetadata`, and it always sets the param `checkAllowAutoCreation` to `false` and can not be set manually.
	- Regarding the client API `HttpLookupService.getPartitionedTopicMetadata`, it depends on the HTTP API `PersistentTopics.getPartitionedMetadata`, and it always sets the param `checkAllowAutoCreation` to `true` and can not be set manually.
	- Regarding the client API `BinaryProtoLookupService.getPartitionedTopicMetadata`, it always tries to create partitioned metadata.
	- `REST API & HTTP API`: Since there are only two implementations of the 4 ways to get partitioned metadata, we call HTTP API `PersistentTopics.getPartitionedMetadata`, `pulsar-admin`, and `HttpLookupService.getPartitionedTopicMetadata` HTTP API, and call `BinaryProtoLookupService.getPartitionedTopicMetadata` Binary API.

	# Motivation

	The param `create if not exists` of the Binary API is always `true.`

	- For case 4 of `pulsarClient.getPartitionsForTopic`'s behavior, it always tries to create the partitioned metadata, but the API name is `getxxx`.
	- For case 5 of `pulsarClient.getPartitionsForTopic`'s behavior, it returns a `0` partitioned metadata, but the topic does not exist. For the correct behavior of this case, we had discussed [here](https://github.com/apache/pulsar/issues/8813) before.
	- BTW, [flink-connector-pulsar](https://github.com/apache/flink-connector-pulsar/blob/main/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java#L221-L227) is using this API to create partitioned topic metadata.

	# Goals

	- Regarding the case 4: Add a new API `PulsarClient.getPartitionsForTopic(String, boolean)` to support the feature that just get partitioned topic metadata and do not try to create one. See detail below.
	- Regarding the case 5: Instead of returning a `0` partitioned metadata, respond to a not found error when calling `pulsarClient.getPartitionsForTopic(String)` if the topic does not exist.

	# Detailed Design

	## Public-facing Changes

	When you call the public API `pulsarClient.getPartitionsForTopic`, pulsar will not create the partitioned metadata anymore.

	### Public API
	LookupService.java
	```java

	- CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(TopicName topicName);

	/**
	* 1. Get the partitions if the topic exists. Return "{partition: n}" if a partitioned topic exists; return "{partition: 0}" if a non-partitioned topic exists.
	* 2. When {@param createIfAutoCreationEnabled} is "false," neither partitioned topic nor non-partitioned topic does not exist. You will get an {@link PulsarClientException.NotFoundException}.
	* 2-1. You will get a {@link PulsarClientException.NotSupportedException} if the broker's version is an older one that does not support this feature and the Pulsar client is using a binary protocol "serviceUrl".
	* 3. When {@param createIfAutoCreationEnabled} is "true," it will trigger an auto-creation for this topic(using the default topic auto-creation strategy you set for the broker), and the corresponding result is returned. For the result, see case 1.
	* @version 3.3.0
	*/
	+ CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(TopicName topicName, boolean createIfAutoCreationEnabled);
	```

	The behavior of the new API `LookupService.getPartitionedTopicMetadata(TopicName, boolean)`.

	\| case \| client-side param: `createIfAutoCreationEnabled` \| non-partitioned topic \| partitioned topic \| broker-side: topic auto-creation strategy \| current behavior \|
	\|------\|--------------------------------------------------\|-----------------------\|-------------------------------------\|--------------------------------------------------------------------------------------------------------------------\|---------------------------------------------------------------------------------------------\|
	\| 1 \| `true/false` \| `exists: true` \| \| \| REST/Binary API: `{partitions: 0}` \|
	\| 2 \| `true/false` \| \| `exists: true` <br> `partitions: 2` \| \| REST/Binary API: `{partitions: 2}` \|
	\| 3 \| `true` \| \| \| `allowAutoTopicCreation`: `true` <be> `allowAutoTopicCreationType`: `non-partitioned` \| REST/Binary API: <br>   - `create new: true` <br>   - `{partitions: 0}` \|
	\| 4 \| `true` \| \| \| `allowAutoTopicCreation`: `true` <be> `allowAutoTopicCreationType`: `partitioned` <br> `defaultNumPartitions`: `2` \| REST/Binary API: <br>   - `create new: true` <br>   - `{partitions: 2}` \|
	\| 5 \| `false` \| \| \| `allowAutoTopicCreation`: `true` \| REST/Binary API: <br>   - Not found error \|
	\| 6 \| `true` \| \| \| `allowAutoTopicCreation`: `false` \| REST/Binary API: <br>   - Not found error \|


	PulsarClient.java
	```java
	// This API existed before. Not change it, thus ensuring compatibility.
	+ @Deprecated it is not suggested to use now; please use {@link #getPartitionsForTopic(TopicName, boolean)}.
	- CompletableFuture<List<String>> getPartitionsForTopic(String topic);
	+ default CompletableFuture<List<String>> getPartitionsForTopic(String topic) {
	+ getPartitionsForTopic(topic, true);
	+ }

	/**
	* 1. Get the partitions if the topic exists. Return "[{partition-0}, {partition-1}....{partition-n}}]" if a partitioned topic exists; return "[{topic}]" if a non-partitioned topic exists.
	* 2. When {@param createIfAutoCreationEnabled} is "false", neither the partitioned topic nor non-partitioned topic does not exist. You will get an {@link PulsarClientException.NotFoundException}.
	* 2-1. You will get a {@link PulsarClientException.NotSupportedException} if the broker's version is an older one that does not support this feature and the Pulsar client is using a binary protocol "serviceUrl".
	* 3. When {@param createIfAutoCreationEnabled} is "true," it will trigger an auto-creation for this topic(using the default topic auto-creation strategy you set for the broker), and the corresponding result is returned. For the result, see case 1.
	* @version 3.3.0
	*/
	CompletableFuture<List<String>> getPartitionsForTopic(String topic, boolean createIfAutoCreationEnabled);
	```

	The behavior of the new API `PulsarClient.getPartitionsForTopic(String, boolean)`.

	\| case \| client-side param: `createIfAutoCreationEnabled` \| non-partitioned topic \| partitioned topic \| broker-side: topic autp-creation strategy \| current behavior \|
	\|------\|--------------------------------------------------\|----------------------\|-------------------------------------\|--------------------------------------------------------------------------------------------------------------------\|-------------------------------------------------------------------------------------------------------------------------------------------------\|
	\| 1 \| `true/false` \| `exists: true` \| \| \| REST/Binary API: `["{tenat}/{ns}/topic"]` \|
	\| 2 \| `true/false` \| \| `exists: true` <br> `partitions: 2` \| \| REST/Binary `API`: `["{tenat}/{ns}/topic-partition-0", "{tenat}/{ns}/topic-partition-1"]` \|
	\| 3 \| `true` \| \| \| `allowAutoTopicCreation`: `true` <be> `allowAutoTopicCreationType`: `non-partitioned` \| REST/Binary API: <br>   - `create new: true` <br>   - `["{tenat}/{ns}/topic"]` \|
	\| 4 \| `true` \| \| \| `allowAutoTopicCreation`: `true` <be> `allowAutoTopicCreationType`: `partitioned` <br> `defaultNumPartitions`: `2` \| REST/Binary API: <br>   - `create new: true` <br>   - `["{tenat}/{ns}/topic-partition-0", "{tenat}/{ns}/topic-partition-1"]` \|
	\| 5 \| `false` \| \| \| `allowAutoTopicCreation`: `true` \| REST/Binary API: <br>   - Not found error \|
	\| 5 \| `true` \| \| \| `allowAutoTopicCreation`: `false` \| REST/Binary API: <br>   - Not found error \|



	### Binary protocol

	CommandPartitionedTopicMetadata
	```
	message CommandPartitionedTopicMetadata {
	+ optional bool metadata_auto_creation_enabled = 6 [default = true];
	}
	```

	FeatureFlags
	```
	message FeatureFlags {
	+ optional bool supports_binary_api_get_partitioned_meta_with_param_created_false = 5 [default = false];
	}
	```

	# Backward & Forward Compatibility

	- Old version client and New version Broker: The client will call the old API.

	- New version client and Old version Broker: The feature flag `supports_binary_api_get_partitioned_meta_with_param_created_false` will be `false`. The client will get a not-support error if the param `createIfAutoCreationEnabled` is false.