You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by si...@apache.org on 2019/01/20 03:11:56 UTC
[pulsar] branch master updated:
create_topic_with_check_namespaces_cluster (#3365)
This is an automated email from the ASF dual-hosted git repository.
sijie 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 c8b274a create_topic_with_check_namespaces_cluster (#3365)
c8b274a is described below
commit c8b274a223b7bfdcbefd21748c81affa8b178eaa
Author: Samuel <fo...@yahoo.com>
AuthorDate: Sun Jan 20 11:11:52 2019 +0800
create_topic_with_check_namespaces_cluster (#3365)
### Motivation
add check namespaces already have the cluster assigned when create topic.
**the old way**:
creating topic will not check if namespaces have broker cluster assigned, but after you created topic successfully , and ran **./pulsar-admin topics list tenant/namespace**, it will prompt **Namespace does not have any clusters configured : local_cluster=** and show nothing.
**the good way**:
check the namespace has already assigned to broker cluster when you are trying to create the topic.
### Modifications
NonPersistentTopics.java and PersistentTopics.java for admin v2 interface
---
.../apache/pulsar/broker/admin/AdminResource.java | 13 +++++++++++
.../broker/admin/v2/NonPersistentTopics.java | 5 ++++-
.../pulsar/broker/admin/v2/PersistentTopics.java | 4 +++-
.../apache/pulsar/broker/admin/AdminApiTest2.java | 25 ++++++++++++++++++++++
4 files changed, 45 insertions(+), 2 deletions(-)
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 b47fda8..35d2286 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
@@ -243,6 +243,19 @@ public abstract class AdminResource extends PulsarWebResource {
}
}
+ protected void validateGlobalNamespaceOwnership(String property, String namespace) {
+ try {
+ this.namespaceName = NamespaceName.get(property, namespace);
+ validateGlobalNamespaceOwnership(this.namespaceName);
+ } catch (IllegalArgumentException e) {
+ throw new RestException(Status.PRECONDITION_FAILED, "Tenant name or namespace is not valid");
+ } catch (RestException re) {
+ throw new RestException(Status.PRECONDITION_FAILED, "Namespace does not have any clusters configured");
+ } catch (Exception e) {
+ log.warn("Failed to validate global cluster configuration : ns={} emsg={}", namespace, e.getMessage());
+ throw new RestException(Status.SERVICE_UNAVAILABLE, "Failed to validate global cluster configuration");
+ }
+ }
@Deprecated
protected void validateNamespaceName(String property, String cluster, String namespace) {
try {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
index 9d49ad1..b8c0d04 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
@@ -107,10 +107,13 @@ public class NonPersistentTopics extends PersistentTopics {
@Path("/{tenant}/{namespace}/{topic}/partitions")
@ApiOperation(value = "Create a partitioned topic.", notes = "It needs to be called before creating a producer on a partitioned topic.")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
- @ApiResponse(code = 409, message = "Partitioned topic already exists") })
+ @ApiResponse(code = 409, message = "Partitioned topic already exists"),
+ @ApiResponse(code = 412, message = "Failed Reason : Name is invalid or Namespace does not have any clusters configured"),
+ @ApiResponse(code = 503, message = "Failed to validate global cluster configuration")})
public void createPartitionedTopic(@PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
@PathParam("topic") @Encoded String encodedTopic, int numPartitions,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateGlobalNamespaceOwnership(tenant,namespace);
validateTopicName(tenant, namespace, encodedTopic);
validateAdminAccessForTenant(topicName.getTenant());
if (numPartitions <= 1) {
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 350da17..30cb4b3 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
@@ -129,11 +129,13 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 409, message = "Partitioned topic already exist"),
- @ApiResponse(code = 412, message = "Partitioned topic name is invalid")
+ @ApiResponse(code = 412, message = "Failed Reason : Name is invalid or Namespace does not have any clusters configured"),
+ @ApiResponse(code = 503, message = "Failed to validate global cluster configuration")
})
public void createPartitionedTopic(@PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
@PathParam("topic") @Encoded String encodedTopic, int numPartitions,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateGlobalNamespaceOwnership(tenant,namespace);
validatePartitionedTopicName(tenant, namespace, encodedTopic);
internalCreatePartitionedTopic(numPartitions, authoritative);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
index 3f90392..066cb66 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -878,4 +878,29 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
// Global cluster, if there, should be omitted from the results
assertEquals(admin.clusters().getClusters(), Lists.newArrayList(cluster));
}
+ /**
+ * verifies cluster has been set before create topic
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testClusterIsReadyBeforeCreateTopic() throws PulsarAdminException {
+ final String topicName = "partitionedTopic";
+ final int partitions = 4;
+ final String persistentPartitionedTopicName = "persistent://prop-xyz/ns2/" + topicName;
+ final String NonPersistentPartitionedTopicName = "non-persistent://prop-xyz/ns2/" + topicName;
+
+ // init tenant and namespace without cluster
+ admin.namespaces().createNamespace("prop-xyz/ns2");
+ try {
+ admin.topics().createPartitionedTopic(persistentPartitionedTopicName, partitions);
+ Assert.fail("should have failed due to Namespace does not have any clusters configured");
+ } catch (PulsarAdminException.PreconditionFailedException e) {
+ }
+ try {
+ admin.topics().createPartitionedTopic(NonPersistentPartitionedTopicName, partitions);
+ Assert.fail("should have failed due to Namespace does not have any clusters configured");
+ } catch (PulsarAdminException.PreconditionFailedException e) {
+ }
+ }
}