fix get partition metadata problem for a non-existed topic (#8818)
Fixes #8813
Currently, getting the partition metadata for a non-existed topic, it returns 0 instead of throwing an exception.
This pr fix this by throwing an exception.
If no metadata found in global zk, then will check whether the topic is exist, if yes, will return 0, otherwise will throw an exception.
(cherry picked from commit a3dfb2a40979a2b9087b078e78044b91be558a76)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index 028c624..86a2f64 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -50,6 +50,7 @@
import org.apache.pulsar.broker.web.PulsarWebResource;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.common.api.proto.PulsarApi;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.naming.Constants;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
@@ -371,6 +372,26 @@
}
}
+ protected void validateTopicExistedAndCheckAllowAutoCreation(String tenant, String namespace,
+ String encodedTopic, boolean checkAllowAutoCreation) {
+ try {
+ PartitionedTopicMetadata partitionedTopicMetadata =
+ pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).get();
+ if (partitionedTopicMetadata.partitions < 1) {
+ if (!pulsar().getNamespaceService().checkTopicExists(topicName).get()
+ && checkAllowAutoCreation
+ && !pulsar().getBrokerService().isAllowAutoTopicCreation(topicName)) {
+ throw new RestException(Status.NOT_FOUND,
+ new PulsarClientException.NotFoundException("Topic not exist"));
+ }
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("Failed to validate topic existed {}://{}/{}/{}",
+ domain(), tenant, namespace, topicName, e);
+ throw new RestException(Status.INTERNAL_SERVER_ERROR, "Check topic partition meta failed.");
+ }
+ }
+
@Deprecated
protected void validateTopicName(String property, String cluster, String namespace, String encodedTopic) {
String topic = Codec.decode(encodedTopic);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
index 80067d1..e2cbe5a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
@@ -345,6 +345,7 @@
@ApiParam(value = "Is check configuration required to automatically create topic")
@QueryParam("checkAllowAutoCreation") @DefaultValue("false") boolean checkAllowAutoCreation) {
validateTopicName(tenant, namespace, encodedTopic);
+ validateTopicExistedAndCheckAllowAutoCreation(tenant, namespace, encodedTopic, checkAllowAutoCreation);
return internalGetPartitionedMetadata(authoritative, checkAllowAutoCreation);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestException.java
index f552333..67d4add 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestException.java
@@ -34,6 +34,7 @@
*/
@SuppressWarnings("serial")
public class RestException extends WebApplicationException {
+ private Throwable cause = null;
static String getExceptionData(Throwable t) {
StringWriter writer = new StringWriter();
writer.append("\n --- An unexpected error occurred in the server ---\n\n");
@@ -58,6 +59,16 @@
super(getResponse(t));
}
+ public RestException(Response.Status status, Throwable t) {
+ this(status.getStatusCode(), t.getMessage());
+ this.cause = t;
+ }
+
+ @Override
+ public Throwable getCause() {
+ return cause;
+ }
+
public RestException(PulsarAdminException cae) {
this(cae.getStatusCode(), cae.getHttpError());
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
index 88d2d2a..585c2b3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -851,7 +851,6 @@
assertEquals(admin.topics().getPartitionedTopicMetadata("persistent://prop-xyz/ns1/ds2").partitions,
0);
-
// check the getPartitionedStats for PartitionedTopic returns only partitions metadata, and no partitions info
assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName).partitions,
admin.topics().getPartitionedStats(partitionedTopicName,false).metadata.partitions);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
index 49d4182..9f93d71 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -265,6 +265,10 @@
Assert.assertEquals(persistentTopics
.getPartitionedMetadata(testTenant, testNamespace, nonPartitionTopic, true, false).partitions,
0);
+
+ Assert.assertEquals(persistentTopics
+ .getPartitionedMetadata(testTenant, testNamespace, nonPartitionTopic, true, true).partitions,
+ 0);
}
@Test
@@ -274,6 +278,10 @@
PartitionedTopicMetadata pMetadata = persistentTopics.getPartitionedMetadata(
testTenant, testNamespace, topicName, true, false);
Assert.assertEquals(pMetadata.partitions, 0);
+
+ PartitionedTopicMetadata metadata = persistentTopics.getPartitionedMetadata(
+ testTenant, testNamespace, topicName, true, true);
+ Assert.assertEquals(metadata.partitions, 0);
}
@Test(expectedExceptions = RestException.class)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java
index 45033b5..8285e98 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java
@@ -83,7 +83,7 @@
// passed non persistent topic here since we can not avoid auto creation on non persistent topic now.
Assert.assertNotNull(consumer);
}
- } catch (PulsarClientException.TopicDoesNotExistException e) {
+ } catch (PulsarClientException.TopicDoesNotExistException | PulsarClientException.NotFoundException e) {
//ok
}
}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
index 597e0d5..f7369d8 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
@@ -908,6 +908,8 @@
return new ConsumerAssignException(msg);
} else if (cause instanceof TopicDoesNotExistException) {
return new TopicDoesNotExistException(msg);
+ } else if (cause instanceof NotFoundException) {
+ return new NotFoundException(msg);
} else {
return new PulsarClientException(t);
}
