You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/10/08 10:18:14 UTC
(pulsar) branch branch-3.3 updated: [fix][client] Make auto partitions update work for old brokers without PIP-344 (#24822)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new 6079a523382 [fix][client] Make auto partitions update work for old brokers without PIP-344 (#24822)
6079a523382 is described below
commit 6079a5233825affee30c66608ed4a66fc8a1f1bc
Author: Yunze Xu <xy...@163.com>
AuthorDate: Wed Oct 8 17:56:40 2025 +0800
[fix][client] Make auto partitions update work for old brokers without PIP-344 (#24822)
(cherry picked from commit 5e59d0e164d5c255021302394d9ad70583fa565a)
---
pip/pip-344.md | 7 ++-
.../apache/pulsar/client/impl/ClientCnxTest.java | 2 +-
.../pulsar/client/impl/PulsarClientImpl.java | 2 +-
.../backwardscompatibility/ClientTest2_2.java | 54 ++++++++++++++++++++++
4 files changed, 61 insertions(+), 4 deletions(-)
diff --git a/pip/pip-344.md b/pip/pip-344.md
index 5eafc6fd5c2..120a8e1ef0d 100644
--- a/pip/pip-344.md
+++ b/pip/pip-344.md
@@ -122,6 +122,9 @@ message FeatureFlags {
# Backward & Forward Compatibility
-- Old version client and New version Broker: The client will call the old API.
+Old version (`< 3.0.6`) client and New version (`>= 3.0.6`) 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.
+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 in the following cases:
+- The topic is a DLQ topic
+- The topic is non-persistent
+- The topic is in geo-replication that the local cluster is new version and the remote cluster is old version
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
index 5fc19f04a39..ffcffdbf207 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
@@ -157,7 +157,7 @@ public class ClientCnxTest extends MockedPulsarServiceBaseTest {
field.set(clientCnxFuture.get(), false);
}
try {
- clientWitBinaryLookup.getPartitionsForTopic(topic, false).join();
+ clientWitBinaryLookup.getPartitionedTopicMetadata(topic, false, false).join();
Assert.fail("Expected an error that the broker version is too old.");
} catch (Exception ex) {
Assert.assertTrue(ex.getMessage().contains("without auto-creation is not supported by the broker"));
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index c8233bd5f9e..5de5f677a2d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -1232,7 +1232,7 @@ public class PulsarClientImpl implements PulsarClient {
@Override
public CompletableFuture<List<String>> getPartitionsForTopic(String topic, boolean metadataAutoCreationEnabled) {
- return getPartitionedTopicMetadata(topic, metadataAutoCreationEnabled, false).thenApply(metadata -> {
+ return getPartitionedTopicMetadata(topic, metadataAutoCreationEnabled, true).thenApply(metadata -> {
if (metadata.partitions > 0) {
TopicName topicName = TopicName.get(topic);
List<String> partitions = new ArrayList<>(metadata.partitions);
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_2.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_2.java
index 212ac48b638..e4e32fa89b1 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_2.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_2.java
@@ -19,8 +19,19 @@
package org.apache.pulsar.tests.integration.backwardscompatibility;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
+import lombok.Cleanup;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageIdAdv;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.TopicMetadata;
import org.apache.pulsar.tests.integration.topologies.ClientTestBase;
+import org.testng.Assert;
import org.testng.annotations.Test;
public class ClientTest2_2 extends PulsarStandaloneTestSuite2_2 {
@@ -33,4 +44,47 @@ public class ClientTest2_2 extends PulsarStandaloneTestSuite2_2 {
clientTestBase.resetCursorCompatibility(serviceUrl.get(), httpServiceUrl.get(), topicName);
}
+ @Test(timeOut = 20000)
+ public void testAutoPartitionsUpdate() throws Exception {
+ @Cleanup final var pulsarClient = PulsarClient.builder()
+ .serviceUrl(getContainer().getPlainTextServiceUrl())
+ .build();
+ final var topic = "test-auto-part-update";
+ final var topic2 = "dummy-topic";
+ @Cleanup final var admin = PulsarAdmin.builder().serviceHttpUrl(getContainer().getHttpServiceUrl()).build();
+ // Use 2 as the initial partition number because old version broker cannot update partitions on a topic that
+ // has only 1 partition.
+ admin.topics().createPartitionedTopic(topic, 2);
+ admin.topics().createPartitionedTopic(topic2, 2);
+ @Cleanup final var producer = pulsarClient.newProducer().autoUpdatePartitions(true)
+ .autoUpdatePartitionsInterval(1, TimeUnit.SECONDS)
+ .messageRoutingMode(MessageRoutingMode.CustomPartition)
+ .messageRouter(new MessageRouter() {
+ @Override
+ public int choosePartition(Message<?> msg, TopicMetadata metadata) {
+ return metadata.numPartitions() - 1;
+ }
+ })
+ .topic(topic)
+ .create();
+ @Cleanup final var consumer = pulsarClient.newConsumer().autoUpdatePartitions(true)
+ .autoUpdatePartitionsInterval(1, TimeUnit.SECONDS)
+ .topic(topic).subscriptionName("sub")
+ .subscribe();
+ @Cleanup final var multiTopicsConsumer = pulsarClient.newConsumer().autoUpdatePartitions(true)
+ .autoUpdatePartitionsInterval(1, TimeUnit.SECONDS)
+ .topics(List.of(topic, topic2)).subscriptionName("sub-2").subscribe();
+
+ admin.topics().updatePartitionedTopic(topic, 3);
+ Thread.sleep(1500);
+ final var msgId = (MessageIdAdv) producer.send("msg".getBytes());
+ Assert.assertEquals(msgId.getPartitionIndex(), 2);
+
+ final var msg = consumer.receive(3, TimeUnit.SECONDS);
+ Assert.assertNotNull(msg);
+ Assert.assertEquals(((MessageIdAdv) msg.getMessageId()).getPartitionIndex(), 2);
+ final var msg2 = multiTopicsConsumer.receive(3, TimeUnit.SECONDS);
+ Assert.assertNotNull(msg2);
+ Assert.assertEquals(msg2.getMessageId(), msg.getMessageId());
+ }
}