diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml index 672c11da03e68..e200d368027ee 100644 --- a/pulsar-broker/pom.xml +++ b/pulsar-broker/pom.xml @@ -696,7 +696,24 @@ false - org.apache.pulsar.broker.admin.v2 + + org.apache.pulsar.broker.admin.v2.Bookies + org.apache.pulsar.broker.admin.v2.BrokerStats + org.apache.pulsar.broker.admin.v2.Brokers + org.apache.pulsar.broker.admin.v2.Clusters + org.apache.pulsar.broker.admin.v2.Functions + org.apache.pulsar.broker.admin.v2.Namespaces + org.apache.pulsar.broker.admin.v2.NonPersistentTopics + org.apache.pulsar.broker.admin.v2.PersistentTopics + + + org.apache.pulsar.broker.admin.v2.ResourceGroups + org.apache.pulsar.broker.admin.v2.ResourceQuotas + org.apache.pulsar.broker.admin.v2.SchemasResource + org.apache.pulsar.broker.admin.v2.Tenants + org.apache.pulsar.broker.admin.v2.Worker + org.apache.pulsar.broker.admin.v2.WorkerStats + http,https /admin/v2 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 3e8219f08f19d..a5e30d2180738 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 @@ -29,7 +29,6 @@ import java.util.Map; import java.util.Optional; import java.util.Set; -import javax.ws.rs.Consumes; import javax.ws.rs.DELETE; import javax.ws.rs.DefaultValue; import javax.ws.rs.Encoded; @@ -300,49 +299,6 @@ public void createPartitionedTopic( } } - @PUT - @Consumes(PartitionedTopicMetadata.MEDIA_TYPE) - @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 = 307, message = "Current broker doesn't serve the namespace of this topic"), - @ApiResponse(code = 401, message = "Don't have permission to administrate resources on this tenant"), - @ApiResponse(code = 403, message = "Don't have admin permission"), - @ApiResponse(code = 404, message = "Tenant or namespace doesn't exist"), - @ApiResponse(code = 406, message = "The number of partitions should be more than 0 and" - + " less than or equal to maxNumPartitionsPerPartitionedTopic"), - @ApiResponse(code = 409, message = "Partitioned topic already exist"), - @ApiResponse(code = 412, - message = "Failed Reason : Name is invalid or Namespace does not have any clusters configured"), - @ApiResponse(code = 500, message = "Internal server error"), - @ApiResponse(code = 503, message = "Failed to validate global cluster configuration") - }) - public void createPartitionedTopic( - @Suspended final AsyncResponse asyncResponse, - @ApiParam(value = "Specify the tenant", required = true) - @PathParam("tenant") String tenant, - @ApiParam(value = "Specify the namespace", required = true) - @PathParam("namespace") String namespace, - @ApiParam(value = "Specify topic name", required = true) - @PathParam("topic") @Encoded String encodedTopic, - @ApiParam(value = "The metadata for the topic", - required = true, type = "PartitionedTopicMetadata") PartitionedTopicMetadata metadata, - @QueryParam("createLocalTopicOnly") @DefaultValue("false") boolean createLocalTopicOnly) { - try { - validateNamespaceName(tenant, namespace); - validateGlobalNamespaceOwnership(); - validatePartitionedTopicName(tenant, namespace, encodedTopic); - validateTopicPolicyOperation(topicName, PolicyName.PARTITION, PolicyOperation.WRITE); - validateCreateTopic(topicName); - internalCreatePartitionedTopic(asyncResponse, metadata.partitions, createLocalTopicOnly, - metadata.properties); - } catch (Exception e) { - log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e); - resumeAsyncResponseExceptionally(asyncResponse, e); - } - } - @PUT @Path("/{tenant}/{namespace}/{topic}") @ApiOperation(value = "Create a non-partitioned topic.", diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopicsExt.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopicsExt.java new file mode 100644 index 0000000000000..441ea1121cdf3 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopicsExt.java @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.pulsar.broker.admin.v2; + +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; +import io.swagger.annotations.ApiParam; +import io.swagger.annotations.ApiResponse; +import io.swagger.annotations.ApiResponses; +import javax.ws.rs.Consumes; +import javax.ws.rs.DefaultValue; +import javax.ws.rs.Encoded; +import javax.ws.rs.PUT; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; +import javax.ws.rs.container.AsyncResponse; +import javax.ws.rs.container.Suspended; +import javax.ws.rs.core.MediaType; +import org.apache.pulsar.broker.admin.impl.PersistentTopicsBase; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; +import org.apache.pulsar.common.policies.data.PolicyName; +import org.apache.pulsar.common.policies.data.PolicyOperation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * This class is for preventing docs conflict before we find a good way to fix + * ISSUE-18947. + */ +@Path("/persistent") +@Produces(MediaType.APPLICATION_JSON) +@Api(value = "/persistent", description = "Persistent topic admin apis", tags = "persistent topic") +public class PersistentTopicsExt extends PersistentTopicsBase { + + @PUT + @Consumes(PartitionedTopicMetadata.MEDIA_TYPE) + @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 = 307, message = "Current broker doesn't serve the namespace of this topic"), + @ApiResponse(code = 401, message = "Don't have permission to administrate resources on this tenant"), + @ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Tenant or namespace doesn't exist"), + @ApiResponse(code = 406, message = "The number of partitions should be more than 0 and" + + " less than or equal to maxNumPartitionsPerPartitionedTopic"), + @ApiResponse(code = 409, message = "Partitioned topic already exist"), + @ApiResponse(code = 412, + message = "Failed Reason : Name is invalid or Namespace does not have any clusters configured"), + @ApiResponse(code = 500, message = "Internal server error"), + @ApiResponse(code = 503, message = "Failed to validate global cluster configuration") + }) + public void createPartitionedTopic( + @Suspended final AsyncResponse asyncResponse, + @ApiParam(value = "Specify the tenant", required = true) + @PathParam("tenant") String tenant, + @ApiParam(value = "Specify the namespace", required = true) + @PathParam("namespace") String namespace, + @ApiParam(value = "Specify topic name", required = true) + @PathParam("topic") @Encoded String encodedTopic, + @ApiParam(value = "The metadata for the topic", + required = true, type = "PartitionedTopicMetadata") PartitionedTopicMetadata metadata, + @QueryParam("createLocalTopicOnly") @DefaultValue("false") boolean createLocalTopicOnly) { + try { + validateNamespaceName(tenant, namespace); + validateGlobalNamespaceOwnership(); + validatePartitionedTopicName(tenant, namespace, encodedTopic); + validateTopicPolicyOperation(topicName, PolicyName.PARTITION, PolicyOperation.WRITE); + validateCreateTopic(topicName); + internalCreatePartitionedTopic(asyncResponse, metadata.partitions, createLocalTopicOnly, + metadata.properties); + } catch (Exception e) { + log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e); + resumeAsyncResponseExceptionally(asyncResponse, e); + } + } + + private static final Logger log = LoggerFactory.getLogger(PersistentTopics.class); +} 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 d25bbc7bd07a9..eb2c878b5500c 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 @@ -16,6 +16,7 @@ * specific language governing permissions and limitations * under the License. */ + package org.apache.pulsar.broker.admin; import static org.mockito.ArgumentMatchers.any; @@ -52,6 +53,7 @@ import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.admin.v2.NonPersistentTopics; import org.apache.pulsar.broker.admin.v2.PersistentTopics; +import org.apache.pulsar.broker.admin.v2.PersistentTopicsExt; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; import org.apache.pulsar.broker.namespace.NamespaceService; @@ -104,6 +106,7 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest { private PersistentTopics persistentTopics; + private PersistentTopicsExt persistentTopicsExt; private final String testTenant = "my-tenant"; private final String testLocalCluster = "use"; private final String testNamespace = "my-namespace"; @@ -134,6 +137,16 @@ protected void setup() throws Exception { doNothing().when(persistentTopics).validateAdminAccessForTenant(this.testTenant); doReturn(mock(AuthenticationDataHttps.class)).when(persistentTopics).clientAuthData(); + persistentTopicsExt = spy(PersistentTopicsExt.class); + persistentTopicsExt.setServletContext(new MockServletContext()); + persistentTopicsExt.setPulsar(pulsar); + doReturn(false).when(persistentTopicsExt).isRequestHttps(); + doReturn(null).when(persistentTopicsExt).originalPrincipal(); + doReturn("test").when(persistentTopicsExt).clientAppId(); + doReturn(TopicDomain.persistent.value()).when(persistentTopicsExt).domain(); + doNothing().when(persistentTopicsExt).validateAdminAccessForTenant(this.testTenant); + doReturn(mock(AuthenticationDataHttps.class)).when(persistentTopicsExt).clientAuthData(); + nonPersistentTopic = spy(NonPersistentTopics.class); nonPersistentTopic.setServletContext(new MockServletContext()); nonPersistentTopic.setPulsar(pulsar); @@ -542,7 +555,7 @@ public void testCreatePartitionedTopic() { Map topicMetadata = new HashMap<>(); topicMetadata.put("key1", "value1"); PartitionedTopicMetadata metadata = new PartitionedTopicMetadata(2, topicMetadata); - persistentTopics.createPartitionedTopic(response2, testTenant, testNamespace, topicName2, metadata, true); + persistentTopicsExt.createPartitionedTopic(response2, testTenant, testNamespace, topicName2, metadata, true); Awaitility.await().untilAsserted(() -> { persistentTopics.getPartitionedMetadata(response2, testTenant, testNamespace, topicName2, true, false); @@ -647,7 +660,7 @@ public void testUpdatePartitionedTopicHavingProperties() throws Exception { ArgumentCaptor responseCaptor = ArgumentCaptor.forClass(PartitionedTopicMetadata.class); PartitionedTopicMetadata metadata = new PartitionedTopicMetadata(2, topicMetadata); - persistentTopics.createPartitionedTopic(response, tenant, namespace, topic, metadata, true); + persistentTopicsExt.createPartitionedTopic(response, tenant, namespace, topic, metadata, true); Awaitility.await().untilAsserted(() -> { persistentTopics.getPartitionedMetadata(response, tenant, namespace, topic, true, false);