[improve] [pip] PIP-344 Correct the behavior of the public API pulsarClient.getPartitionsForTopic(topicName)

pip/pip-344.md

@@ -0,0 +1,78 @@
+# 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> Client API: `partitions: 0` |
+| 2 | `true/false` | `true/false` | | `exists: true` <br> `partitions: 3` | REST API: `partitions: 3`<br> Client API: `partitions: 3` |
+| 3 | `true` | `true` | | | REST API: <br> &nbsp;&nbsp;- `create new: true` <br> &nbsp;&nbsp;- `partitions: 3` <br> Client 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> Client API: <br> &nbsp;&nbsp;not support <br> |
+| 5 | `false` | `true` | | | REST API: <br> &nbsp;&nbsp;- `create new: false` <br> &nbsp;&nbsp;- `partitions: 0` <br> Client API: <br> &nbsp;&nbsp;- `create new: false` <br> &nbsp;&nbsp;- `partitions: 0` <br> |
+
+# Motivation
+
+The param `create if not exists` of the Client 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.
+
+# Goals
+
+Correct the behaviors of case 4 and case 5.
+
+- Do not create the partitioned metadata when calling `pulsarClient.getPartitionsForTopic`. The partitioned metadata will only be created when consumers/producers are trying to register.
+- Instead of returning a `0` partitioned metadata, respond to a not found error when calling `pulsarClient.getPartitionsForTopic` 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.
+- [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.
+
+### Public API
+**LookupService.java**
+```
+// This API existed before. Not change it, thus ensuring compatibility.
+CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(TopicName topicName);
+
++ // A new API that contains an additional param "createIfAutoCreationEnabled."
++ CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(TopicName topicName, boolean createIfAutoCreationEnabled);
+```
+
+| case | do create partitioned metadata | non-partitioned topic | partitioned topic |  current behavior |
+| --- | --- | --- | --- | --- |
+| 1 | `true/false` | `exists: true` | | REST/Client API: `partitions: 0` |
+| 2 | `true/false` | | `exists: true` <br> `partitions: 3` | REST/Client API: `partitions: 3` |
+| 3 | `true` | | | REST/Client API: <br> &nbsp;&nbsp;- `create new: true` <br> &nbsp;&nbsp;- `partitions: 3` |
+| 4 | `false` | | | REST/Client API: <br> &nbsp;&nbsp;- Not found error |
+
+### Binary protocol
+
+**CommandPartitionedTopicMetadata**
+```
+message CommandPartitionedTopicMetadata {
+  + optional bool metadata_auto_creation_enabled = 6 [default = true];
+}
+```
+
+# Backward & Forward Compatibility
+
+- Old version client and New version Broker
+  - Pulsar will create the partitioned topic metadata when calling `pulsarClient.getPartitionsForTopic`.
+  - The client does not pass the `create_if_auto_creation_enabled` parameter of `CommandPartitionedTopicMetadata` to the Broker, which defaults to `true` and automatically creates the partitioned topic metadata.
+
+- New version client and Old version Broker
+  - Pulsar will create the partitioned topic metadata when calling `pulsarClient.getPartitionsForTopic`.
+  - The client passes the `create_if_auto_creation_enabled` parameter of `CommandPartitionedTopicMetadata` to the Broker; the Broker will discard it and create the partitioned topic metadata automatically.
+
+- New version client and New version Broker
+  - Pulsar will not create the partitioned topic metadata when calling `pulsarClient.getPartitionsForTopic`.