This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new afe4261e2b4 [improve] [pip] PIP-344 Correct the behavior of the public
API pulsarClient.getPartitionsForTopic(topicName) (#22182)
afe4261e2b4 is described below
commit afe4261e2b4c07df5498649d617f76e263ab1119
Author: fengyubiao <[email protected]>
AuthorDate: Mon Mar 25 15:20:48 2024 +0800
[improve] [pip] PIP-344 Correct the behavior of the public API
pulsarClient.getPartitionsForTopic(topicName) (#22182)
---
pip/pip-344.md | 127 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 127 insertions(+)
diff --git a/pip/pip-344.md b/pip/pip-344.md
new file mode 100644
index 00000000000..5eafc6fd5c2
--- /dev/null
+++ b/pip/pip-344.md
@@ -0,0 +1,127 @@
+# 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.
