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 2022/04/24 13:37:29 UTC
[GitHub] [pulsar] liangyepianzhou opened a new pull request, #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
liangyepianzhou opened a new pull request, #15296:
URL: https://github.com/apache/pulsar/pull/15296
### Motivation & Modification
Add admin api updateTransactionCoordinatorNumber to change the partition number of tc
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### 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): (yes / no)
- The public API: (yes / no)
- The schema: (yes / no / don't know)
- The default values of configurations: (yes / no)
- The wire protocol: (yes / no)
- The rest endpoints: (yes / no)
- The admin cli options: (yes / no)
- Anything that affects deployment: (yes / no / don't know)
### Documentation
Check the box below or label this PR directly.
Need to update docs?
- [ ] `doc-required`
(Your PR needs to update docs and you will update later)
- [x] `no-need-doc`
(Please explain why)
- [ ] `doc`
(Your PR contains doc changes)
- [ ] `doc-added`
(Docs have been already added)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r859327112
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,27 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(AsyncResponse asyncResponse, int replicas) {
Review Comment:
We should define this method as `protected CompletableFuture<Void> internalScaleTransactionCoordinators`,
which means we can't pass `AsyncResponse` into the method.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r860337638
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java:
##########
@@ -314,4 +316,25 @@ public void getPendingAckInternalStats(@Suspended final AsyncResponse asyncRespo
resumeAsyncResponseExceptionally(asyncResponse, ex);
}
}
+
+ @POST
+ @Path("/transactionCoordinator/replicas")
+ @ApiResponses(value = {
+ @ApiResponse(code = 503, message = "This Broker is not configured "
+ + "with transactionCoordinatorEnabled=true."),
Review Comment:
Have you added this check in the implementation?
And should also add
```
@ApiResponse(code = 401, message = "This operation requires super-user access"),
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,24 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected CompletableFuture<Void> internalScaleTransactionCoordinators(int replicas) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
Review Comment:
We don't need to create the CompletableFuture here but complete it in another thread.
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTransactions.java:
##########
@@ -189,5 +200,6 @@ public CmdTransactions(Supplier<PulsarAdmin> admin) {
jcommander.addCommand("transaction-in-pending-ack-stats", new GetTransactionInPendingAckStats());
jcommander.addCommand("transaction-metadata", new GetTransactionMetadata());
jcommander.addCommand("slow-transactions", new GetSlowTransactions());
+ jcommander.addCommand("scale-transactionCoordinators", new ScaleTransactionCoordinators());
Review Comment:
Need a test to cover here to make sure the CLI works.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,24 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected CompletableFuture<Void> internalScaleTransactionCoordinators(int replicas) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ validateSuperUserAccessAsync()
+ .thenCompose((ignore)-> namespaceResources().getPartitionedTopicResources()
+ .updatePartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, p -> {
+ if (p.partitions >= replicas) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of transaction coordinators should "
+ + "be more than the current number of transaction coordinator");
+ }
+ return new PartitionedTopicMetadata(replicas);
+ }).thenAccept(r -> completableFuture.complete(null)))
+ .exceptionally(e -> {
+ log.error("{} Failed to update the scale of transaction coordinators", clientAppId());
+ completableFuture.completeExceptionally(e);
+ return null;
Review Comment:
Method `scaleTransactionCoordinators` already handled the exceptions, we don't need to handle them again here. Just return the `CompletableFuture`.
And for the log here, we don't need an error log, warn level log is enough.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,24 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected CompletableFuture<Void> internalScaleTransactionCoordinators(int replicas) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ validateSuperUserAccessAsync()
+ .thenCompose((ignore)-> namespaceResources().getPartitionedTopicResources()
+ .updatePartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, p -> {
+ if (p.partitions >= replicas) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of transaction coordinators should "
+ + "be more than the current number of transaction coordinator");
+ }
+ return new PartitionedTopicMetadata(replicas);
+ }).thenAccept(r -> completableFuture.complete(null)))
Review Comment:
We can remove the `.thenAccept(r -> completableFuture.complete(null)))`, just return a CompletableFuture in `updatePartitionedTopicAsync`?
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java:
##########
@@ -519,6 +519,38 @@ public void testTransactionNotEnabled() throws Exception {
} catch (PulsarAdminException ex) {
assertEquals(ex.getStatusCode(), HttpStatus.SC_SERVICE_UNAVAILABLE);
}
+ try {
+ admin.transactions().scaleTransactionCoordinators(1);
+ } catch (PulsarAdminException ex) {
+ assertEquals(ex.getStatusCode(), HttpStatus.SC_SERVICE_UNAVAILABLE);
+ }
+ }
+
+ @Test
+ public void testUpdateTransactionCoordinatorNumber() throws Exception {
+ int coordinatorSize = 3;
+ pulsar.getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources()
+ .createPartitionedTopic(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN,
+ new PartitionedTopicMetadata(coordinatorSize));
+ try {
+ admin.transactions().scaleTransactionCoordinators(coordinatorSize - 1);
Review Comment:
Could you please also add a test for auth enabled? only the super user can access this API.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] liangyepianzhou commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857541845
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTransactions.java:
##########
@@ -178,6 +178,17 @@ void run() throws Exception {
}
}
+ @Parameters(commandDescription = "Update the scale of transaction coordinators")
+ private class ScaleTransactionCoordinators extends CliCommand {
+ @Parameter(names = { "-r", "--replicas" }, description = "The scale of the transaction coordinators")
+ private int replicas;
Review Comment:
Maybe we don't need to tangle here, I have followed Penghui's suggestion to change from `partitions` to `replicas`.
Both names are fine. How about to keep it like this?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] liangyepianzhou commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857545195
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +434,35 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(int replicas, boolean authoritative) {
+ final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
+ if (maxPartitions > 0 && replicas > maxPartitions) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of partitions should be less than or equal to " + maxPartitions);
+ }
+ validateTopicOwnership(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, authoritative);
+ CompletableFuture<Void> updatePartition = new CompletableFuture<>();
+ namespaceResources().getPartitionedTopicResources()
+ .updatePartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, p -> {
+ if (p.partitions >= replicas) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE, "Number of partitions should "
+ + "be more than the current number of transaction coordinator partitions");
+ }
+ return new PartitionedTopicMetadata(replicas);
+ }).thenAccept(r -> updatePartition.complete(null)).exceptionally(ex -> {
+ updatePartition.completeExceptionally(ex.getCause());
+ return null;
+ });
+ try {
+ updatePartition.get(DEFAULT_OPERATION_TIMEOUT_SEC, TimeUnit.SECONDS);
Review Comment:
OK, I will abide by this specification
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r859293737
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,32 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(AsyncResponse asyncResponse, int replicas) {
+ final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
+ if (maxPartitions > 0 && replicas > maxPartitions) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of partitions should be less than or equal to " + maxPartitions);
+ }
+ validateSuperUserAccessAsync().thenAccept((ignore) -> {
+ namespaceResources().getPartitionedTopicResources()
+ .updatePartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, p -> {
+ if (p.partitions >= replicas) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of partitions should "
Review Comment:
```suggestion
"Number of partitions should "
```
Number of transaction coordinators
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,32 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(AsyncResponse asyncResponse, int replicas) {
+ final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
+ if (maxPartitions > 0 && replicas > maxPartitions) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of partitions should be less than or equal to " + maxPartitions);
Review Comment:
may we don't need this config to control
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,32 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(AsyncResponse asyncResponse, int replicas) {
+ final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
+ if (maxPartitions > 0 && replicas > maxPartitions) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of partitions should be less than or equal to " + maxPartitions);
+ }
+ validateSuperUserAccessAsync().thenAccept((ignore) -> {
+ namespaceResources().getPartitionedTopicResources()
+ .updatePartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, p -> {
+ if (p.partitions >= replicas) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of partitions should "
+ + "be more than the current number of transaction coordinator partitions");
Review Comment:
```suggestion
+ "be more than the current number of transaction coordinator partitions");
```
be more than the currenst number of transaction coordinator
##########
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TransactionsImpl.java:
##########
@@ -335,4 +338,17 @@ public TransactionPendingAckInternalStats getPendingAckInternalStats(String topi
return sync(() -> getPendingAckInternalStatsAsync(topic, subName, metadata));
}
+ @Override
+ public void scaleTransactionCoordinators(int replicas) throws PulsarAdminException {
+ sync(() -> scaleTransactionCoordinatorsAsync(replicas));
+ }
+
+ @Override
+ public CompletableFuture<Void> scaleTransactionCoordinatorsAsync(int replicas) {
+ checkArgument(replicas > 0, "Number of partitions must be more than 0");
Review Comment:
transaction coordinators
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] liangyepianzhou commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857460743
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java:
##########
@@ -314,4 +315,17 @@ public void getPendingAckInternalStats(@Suspended final AsyncResponse asyncRespo
resumeAsyncResponseExceptionally(asyncResponse, ex);
}
}
+
+ @POST
+ @Path("/transactionCoordinator/partitions")
+ @ApiResponses(value = {
+ @ApiResponse(code = 503, message = "This Broker is not configured "
+ + "with transactionCoordinatorEnabled=true."),
+ @ApiResponse(code = 406, message = "The number of partitions should be more than "
+ + "the current number of transaction coordinator partitions"
+ + " and less than or equal to maxNumPartitionsPerPartitionedTopic")})
+ public void updateTransactionCoordinatorNumber(int numPartitions) {
+ checkTransactionCoordinatorEnabled();
Review Comment:
I have add the auth just like `updatePartitionedTopic` at last commit. Are there other problems?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857511592
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTransactions.java:
##########
@@ -178,6 +178,17 @@ void run() throws Exception {
}
}
+ @Parameters(commandDescription = "Update the scale of transaction coordinators")
+ private class ScaleTransactionCoordinators extends CliCommand {
+ @Parameter(names = { "-r", "--replicas" }, description = "The scale of the transaction coordinators")
+ private int replicas;
Review Comment:
Which name is better, `partitions` or `replicas` ?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] liangyepianzhou commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857543709
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +434,35 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(int replicas, boolean authoritative) {
+ final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
+ if (maxPartitions > 0 && replicas > maxPartitions) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of partitions should be less than or equal to " + maxPartitions);
+ }
+ validateTopicOwnership(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, authoritative);
Review Comment:
This is for authentication, just like `internalUpdatePartitionedTopic` in `PersistentTopicsBase`.
It should be sync.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857511856
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +434,35 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(int replicas, boolean authoritative) {
+ final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
+ if (maxPartitions > 0 && replicas > maxPartitions) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of partitions should be less than or equal to " + maxPartitions);
+ }
+ validateTopicOwnership(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, authoritative);
Review Comment:
validateTopicOwnership -> validateTopicOwnershipAsync
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857211716
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java:
##########
@@ -314,4 +315,17 @@ public void getPendingAckInternalStats(@Suspended final AsyncResponse asyncRespo
resumeAsyncResponseExceptionally(asyncResponse, ex);
}
}
+
+ @POST
+ @Path("/transactionCoordinator/partitions")
+ @ApiResponses(value = {
+ @ApiResponse(code = 503, message = "This Broker is not configured "
+ + "with transactionCoordinatorEnabled=true."),
+ @ApiResponse(code = 406, message = "The number of partitions should be more than "
+ + "the current number of transaction coordinator partitions"
+ + " and less than or equal to maxNumPartitionsPerPartitionedTopic")})
+ public void updateTransactionCoordinatorNumber(int numPartitions) {
+ checkTransactionCoordinatorEnabled();
Review Comment:
should add auth
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
codelipenghui merged PR #15296:
URL: https://github.com/apache/pulsar/pull/15296
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857512289
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +434,35 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(int replicas, boolean authoritative) {
+ final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
+ if (maxPartitions > 0 && replicas > maxPartitions) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of partitions should be less than or equal to " + maxPartitions);
+ }
+ validateTopicOwnership(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, authoritative);
+ CompletableFuture<Void> updatePartition = new CompletableFuture<>();
+ namespaceResources().getPartitionedTopicResources()
+ .updatePartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, p -> {
+ if (p.partitions >= replicas) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE, "Number of partitions should "
+ + "be more than the current number of transaction coordinator partitions");
+ }
+ return new PartitionedTopicMetadata(replicas);
+ }).thenAccept(r -> updatePartition.complete(null)).exceptionally(ex -> {
+ updatePartition.completeExceptionally(ex.getCause());
+ return null;
+ });
+ try {
+ updatePartition.get(DEFAULT_OPERATION_TIMEOUT_SEC, TimeUnit.SECONDS);
Review Comment:
New REST APIs should all use the async method.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] liangyepianzhou commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857541845
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTransactions.java:
##########
@@ -178,6 +178,17 @@ void run() throws Exception {
}
}
+ @Parameters(commandDescription = "Update the scale of transaction coordinators")
+ private class ScaleTransactionCoordinators extends CliCommand {
+ @Parameter(names = { "-r", "--replicas" }, description = "The scale of the transaction coordinators")
+ private int replicas;
Review Comment:
Maybe we don't need to tangle here, I have followed Penghui's suggestion to change from `partitions` to `replicas`.
Both names are fine. How to keep it like this?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r861441014
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,17 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected CompletableFuture<Void> internalScaleTransactionCoordinators(int replicas) {
+ return validateSuperUserAccessAsync()
+ .thenCompose((ignore)-> namespaceResources().getPartitionedTopicResources()
Review Comment:
(ignore) -> __
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r859593616
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,27 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(AsyncResponse asyncResponse, int replicas) {
+ validateSuperUserAccessAsync().thenAccept((ignore) -> {
+ namespaceResources().getPartitionedTopicResources()
+ .updatePartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, p -> {
+ if (p.partitions >= replicas) {
+ throw new RestException(Response.Status.NOT_ACCEPTABLE,
+ "Number of transaction coordinators should "
+ + "be more than the current number of transaction coordinator");
+ }
+ return new PartitionedTopicMetadata(replicas);
+ }).thenAccept(r -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("{} Failed to update the scale of transaction coordinators", clientAppId());
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
+ }).exceptionally(e -> {
+ log.error("{} Failed to update the scale of transaction coordinators", clientAppId());
+ resumeAsyncResponseExceptionally(asyncResponse, e);
+ return null;
+ });
Review Comment:
Use `thenCompose().thenAccept().exceptionally()` to simplify the implementation.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] liangyepianzhou commented on pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#issuecomment-1110998776
/pulsarbot run-failure-checks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r859592727
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java:
##########
@@ -432,4 +433,27 @@ protected void validateTopicName(String property, String namespace, String encod
throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
}
}
+
+ protected void internalScaleTransactionCoordinators(AsyncResponse asyncResponse, int replicas) {
Review Comment:
Yes, agree. It's a great practice.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] liangyepianzhou commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r860451854
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java:
##########
@@ -314,4 +316,25 @@ public void getPendingAckInternalStats(@Suspended final AsyncResponse asyncRespo
resumeAsyncResponseExceptionally(asyncResponse, ex);
}
}
+
+ @POST
+ @Path("/transactionCoordinator/replicas")
+ @ApiResponses(value = {
+ @ApiResponse(code = 503, message = "This Broker is not configured "
+ + "with transactionCoordinatorEnabled=true."),
Review Comment:
`checkTransactionCoordinatorEnabled(); ` had been aaded, and I added auth and auth test this time.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857208417
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java:
##########
@@ -314,4 +315,17 @@ public void getPendingAckInternalStats(@Suspended final AsyncResponse asyncRespo
resumeAsyncResponseExceptionally(asyncResponse, ex);
}
}
+
+ @POST
+ @Path("/transactionCoordinator/partitions")
+ @ApiResponses(value = {
+ @ApiResponse(code = 503, message = "This Broker is not configured "
+ + "with transactionCoordinatorEnabled=true."),
+ @ApiResponse(code = 406, message = "The number of partitions should be more than "
+ + "the current number of transaction coordinator partitions"
+ + " and less than or equal to maxNumPartitionsPerPartitionedTopic")})
+ public void updateTransactionCoordinatorNumber(int numPartitions) {
Review Comment:
It's better to use `scaleTransactionCoordinators(int replicas)`, it's a more user-facing name, and currently, we can only support scale-up.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #15296: [Improve][txn] Add admin api updateTransactionCoordinatorNumber
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #15296:
URL: https://github.com/apache/pulsar/pull/15296#discussion_r857407532
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java:
##########
@@ -314,4 +315,17 @@ public void getPendingAckInternalStats(@Suspended final AsyncResponse asyncRespo
resumeAsyncResponseExceptionally(asyncResponse, ex);
}
}
+
+ @POST
+ @Path("/transactionCoordinator/partitions")
+ @ApiResponses(value = {
+ @ApiResponse(code = 503, message = "This Broker is not configured "
+ + "with transactionCoordinatorEnabled=true."),
+ @ApiResponse(code = 406, message = "The number of partitions should be more than "
+ + "the current number of transaction coordinator partitions"
+ + " and less than or equal to maxNumPartitionsPerPartitionedTopic")})
+ public void updateTransactionCoordinatorNumber(int numPartitions) {
+ checkTransactionCoordinatorEnabled();
Review Comment:
+1, only the super user can make this change.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org