This is an automated email from the ASF dual-hosted git repository. penghui pushed a commit to branch branch-2.8 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 50da9b228695fcbe19f746152e45da1c67cc5b19 Author: Aloys <[email protected]> AuthorDate: Fri Jun 18 14:09:06 2021 +0800 fix non-persistent topic get partitioned metadata error on discovery (#10806) Fixes #10443 ### Motivation fix non-persistent topic get partitioned metadata error if using discovery (cherry picked from commit 859922942759aaa539fe7b0951a614bb75c71ea8) --- .../discovery/service/BrokerDiscoveryProvider.java | 2 +- .../discovery/service/BaseDiscoveryTestSetup.java | 7 +++++++ .../discovery/service/DiscoveryServiceTest.java | 21 +++++++++++++++++++++ 3 files changed, 29 insertions(+), 1 deletion(-) diff --git a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/BrokerDiscoveryProvider.java b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/BrokerDiscoveryProvider.java index 7f3eb6c..9c9a095 100644 --- a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/BrokerDiscoveryProvider.java +++ b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/BrokerDiscoveryProvider.java @@ -106,7 +106,7 @@ public class BrokerDiscoveryProvider implements Closeable { try { checkAuthorization(service, topicName, role, authenticationData); final String path = path(PARTITIONED_TOPIC_PATH_ZNODE, - topicName.getNamespaceObject().toString(), "persistent", topicName.getEncodedLocalName()); + topicName.getNamespaceObject().toString(), topicName.getDomain().value(), topicName.getEncodedLocalName()); // gets the number of partitions from the zk cache pulsarResources.getNamespaceResources().getPartitionedTopicResources().getAsync(path) .thenAccept(metadata -> { diff --git a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/BaseDiscoveryTestSetup.java b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/BaseDiscoveryTestSetup.java index 8a0bd00..70fc26f 100644 --- a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/BaseDiscoveryTestSetup.java +++ b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/BaseDiscoveryTestSetup.java @@ -80,4 +80,11 @@ public class BaseDiscoveryTestSetup { && path.equals("/admin/partitioned-topics/test/local/ns/persistent/my-topic-2"); }); } + + protected void simulateStoreErrorForNonPersistentTopic(String string, Code sessionexpired) { + mockZooKeeper.failConditional(Code.SESSIONEXPIRED, (op, path) -> { + return op == MockZooKeeper.Op.GET + && path.equals("/admin/partitioned-topics/test/local/ns/non-persistent/my-topic-2"); + }); + } } diff --git a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/DiscoveryServiceTest.java b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/DiscoveryServiceTest.java index 73b3c41..802879f 100644 --- a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/DiscoveryServiceTest.java +++ b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/DiscoveryServiceTest.java @@ -114,6 +114,27 @@ public class DiscoveryServiceTest extends BaseDiscoveryTestSetup { } } + @Test + public void testGetPartitionsMetadataForNonPersistentTopic() throws Exception { + TopicName topic1 = TopicName.get("non-persistent://test/local/ns/my-topic-1"); + + PartitionedTopicMetadata m = service.getDiscoveryProvider().getPartitionedTopicMetadata(service, topic1, "role", null) + .get(); + assertEquals(m.partitions, 0); + + // Simulate ZK error + simulateStoreErrorForNonPersistentTopic("/admin/partitioned-topics/test/local/ns/non-persistent/my-topic-2", Code.SESSIONEXPIRED); + TopicName topic2 = TopicName.get("non-persistent://test/local/ns/my-topic-2"); + CompletableFuture<PartitionedTopicMetadata> future = service.getDiscoveryProvider() + .getPartitionedTopicMetadata(service, topic2, "role", null); + try { + future.get(); + fail("Partition metadata lookup should have failed"); + } catch (ExecutionException e) { + assertEquals(e.getCause().getClass(), MetadataStoreException.class); + } + } + /** * It verifies: client connects to Discovery-service and receives discovery response successfully. *
