You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/03/05 04:01:45 UTC
[GitHub] [pulsar] codelipenghui opened a new pull request #6478: Fix create
partitioned topic with a substring of an existing topic name.
codelipenghui opened a new pull request #6478: Fix create partitioned topic with a substring of an existing topic name.
URL: https://github.com/apache/pulsar/pull/6478
Fixes #6468
### Motivation
Fix create a partitioned topic with a substring of an existing topic name. And make create partitioned topic async.
### Verifying this change
Added unit test for this change.
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: (no)
- The default values of configurations: (no)
- The wire protocol: (no)
- The rest endpoints: (no)
- The admin cli options: (no)
- Anything that affects deployment: (no)
### Documentation
- Does this pull request introduce a new feature? (no)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] jiazhai commented on issue #6478: Fix create partitioned
topic with a substring of an existing topic name.
Posted by GitBox <gi...@apache.org>.
jiazhai commented on issue #6478: Fix create partitioned topic with a substring of an existing topic name.
URL: https://github.com/apache/pulsar/pull/6478#issuecomment-598575011
cherry-picked into branch-2.5
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] codelipenghui merged pull request #6478: Fix create
partitioned topic with a substring of an existing topic name.
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #6478: Fix create partitioned topic with a substring of an existing topic name.
URL: https://github.com/apache/pulsar/pull/6478
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] sijie commented on a change in pull request #6478: Fix
create partitioned topic with a substring of an existing topic name.
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #6478: Fix create partitioned topic with a substring of an existing topic name.
URL: https://github.com/apache/pulsar/pull/6478#discussion_r388655164
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
##########
@@ -707,4 +709,93 @@ protected boolean isNamespaceReplicated(NamespaceName namespaceName) {
partitionedTopics.sort(null);
return partitionedTopics;
}
+
+ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions) {
+ try {
+ validateAdminAccessForTenant(topicName.getTenant());
+ } catch (Exception e) {
+ log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e);
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return;
+ }
+ if (numPartitions <= 0) {
+ asyncResponse.resume(new RestException(Status.NOT_ACCEPTABLE, "Number of partitions should be more than 0"));
+ return;
+ }
+ checkTopicExistsAsync(topicName).thenAccept(exists -> {
+ if (exists) {
+ log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
+ asyncResponse.resume(new RestException(Status.CONFLICT, "This topic already exists"));
+ } else {
+ try {
+ String path = ZkAdminPaths.partitionedTopicPath(topicName);
+ byte[] data = jsonMapper().writeValueAsBytes(new PartitionedTopicMetadata(numPartitions));
+ zkCreateOptimisticAsync(globalZk(), path, data, (rc, s, o, s1) -> {
+ if (KeeperException.Code.OK.intValue() == rc) {
+ if (topicName.isPersistent()) {
+ tryCreatePartitionsAsync(numPartitions);
+ }
+ globalZk().sync(path, (rc2, s2, ctx) -> {
Review comment:
same comment as above
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] codelipenghui commented on a change in pull request #6478:
Fix create partitioned topic with a substring of an existing topic name.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #6478: Fix create partitioned topic with a substring of an existing topic name.
URL: https://github.com/apache/pulsar/pull/6478#discussion_r388673263
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
##########
@@ -707,4 +709,93 @@ protected boolean isNamespaceReplicated(NamespaceName namespaceName) {
partitionedTopics.sort(null);
return partitionedTopics;
}
+
+ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions) {
+ try {
+ validateAdminAccessForTenant(topicName.getTenant());
+ } catch (Exception e) {
+ log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e);
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return;
+ }
+ if (numPartitions <= 0) {
+ asyncResponse.resume(new RestException(Status.NOT_ACCEPTABLE, "Number of partitions should be more than 0"));
+ return;
+ }
+ checkTopicExistsAsync(topicName).thenAccept(exists -> {
+ if (exists) {
+ log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
+ asyncResponse.resume(new RestException(Status.CONFLICT, "This topic already exists"));
+ } else {
+ try {
+ String path = ZkAdminPaths.partitionedTopicPath(topicName);
+ byte[] data = jsonMapper().writeValueAsBytes(new PartitionedTopicMetadata(numPartitions));
+ zkCreateOptimisticAsync(globalZk(), path, data, (rc, s, o, s1) -> {
Review comment:
The partitioned topic metadata is stored in the global zookeeper.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] codelipenghui commented on issue #6478: Fix create
partitioned topic with a substring of an existing topic name.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #6478: Fix create partitioned topic with a substring of an existing topic name.
URL: https://github.com/apache/pulsar/pull/6478#issuecomment-595571085
@sijie I have addressed your comments, please take a look.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] sijie commented on a change in pull request #6478: Fix
create partitioned topic with a substring of an existing topic name.
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #6478: Fix create partitioned topic with a substring of an existing topic name.
URL: https://github.com/apache/pulsar/pull/6478#discussion_r388655914
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
##########
@@ -707,4 +709,93 @@ protected boolean isNamespaceReplicated(NamespaceName namespaceName) {
partitionedTopics.sort(null);
return partitionedTopics;
}
+
+ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions) {
+ try {
+ validateAdminAccessForTenant(topicName.getTenant());
+ } catch (Exception e) {
+ log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e);
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return;
+ }
+ if (numPartitions <= 0) {
+ asyncResponse.resume(new RestException(Status.NOT_ACCEPTABLE, "Number of partitions should be more than 0"));
+ return;
+ }
+ checkTopicExistsAsync(topicName).thenAccept(exists -> {
+ if (exists) {
+ log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
+ asyncResponse.resume(new RestException(Status.CONFLICT, "This topic already exists"));
+ } else {
+ try {
+ String path = ZkAdminPaths.partitionedTopicPath(topicName);
+ byte[] data = jsonMapper().writeValueAsBytes(new PartitionedTopicMetadata(numPartitions));
+ zkCreateOptimisticAsync(globalZk(), path, data, (rc, s, o, s1) -> {
+ if (KeeperException.Code.OK.intValue() == rc) {
+ if (topicName.isPersistent()) {
+ tryCreatePartitionsAsync(numPartitions);
+ }
+ globalZk().sync(path, (rc2, s2, ctx) -> {
Review comment:
This should happen in the callback of `tryCreatePartitionAsync`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [pulsar] sijie commented on a change in pull request #6478: Fix
create partitioned topic with a substring of an existing topic name.
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #6478: Fix create partitioned topic with a substring of an existing topic name.
URL: https://github.com/apache/pulsar/pull/6478#discussion_r388655026
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
##########
@@ -707,4 +709,93 @@ protected boolean isNamespaceReplicated(NamespaceName namespaceName) {
partitionedTopics.sort(null);
return partitionedTopics;
}
+
+ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions) {
+ try {
+ validateAdminAccessForTenant(topicName.getTenant());
+ } catch (Exception e) {
+ log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e);
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return;
+ }
+ if (numPartitions <= 0) {
+ asyncResponse.resume(new RestException(Status.NOT_ACCEPTABLE, "Number of partitions should be more than 0"));
+ return;
+ }
+ checkTopicExistsAsync(topicName).thenAccept(exists -> {
+ if (exists) {
+ log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
+ asyncResponse.resume(new RestException(Status.CONFLICT, "This topic already exists"));
+ } else {
+ try {
+ String path = ZkAdminPaths.partitionedTopicPath(topicName);
+ byte[] data = jsonMapper().writeValueAsBytes(new PartitionedTopicMetadata(numPartitions));
+ zkCreateOptimisticAsync(globalZk(), path, data, (rc, s, o, s1) -> {
Review comment:
@codelipenghui why it is `globalZk()` here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services