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/01/21 06:11:02 UTC
[GitHub] [pulsar] liudezhi2098 opened a new pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
liudezhi2098 opened a new pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878
### Motivation
Avoid call sync method in async rest API for PersistentTopicsBase#internalExpireMessagesByPosition.
### Modifications
- *Use async instead of sync method.*
### 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
Need to update docs?
- [x] `no-need-doc`
--
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] liudezhi2098 commented on pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
Posted by GitBox <gi...@apache.org>.
liudezhi2098 commented on pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878#issuecomment-1022829877
/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] mattisonchao commented on pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878#issuecomment-1018489572
emmm. i think we need to refactor this code. it seems very like callback hell.
@liudezhi2098
Do you have any good idea ?
--
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 change in pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878#discussion_r789553921
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -3447,98 +3447,141 @@ private void internalExpireMessagesByTimestampForSinglePartition(String subName,
protected void internalExpireMessagesByPosition(AsyncResponse asyncResponse, String subName, boolean authoritative,
MessageIdImpl messageId, boolean isExcluded, int batchIndex) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.warn("[{}][{}] Failed to expire messages on subscription {} to position {}: {}", clientAppId(),
- topicName, subName, messageId, e.getMessage());
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- }
-
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.EXPIRE_MESSAGES);
-
- log.info("[{}][{}] received expire messages on subscription {} to position {}", clientAppId(), topicName,
- subName, messageId);
-
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (!topicName.isPartitioned() && getPartitionedTopicMetadata(topicName, authoritative, false).partitions > 0) {
- log.warn("[{}] Not supported operation expire message up to {} on partitioned-topic {} {}",
- clientAppId(), messageId, topicName, subName);
- asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
- "Expire message at position is not supported for partitioned-topic"));
- return;
- } else if (messageId.getPartitionIndex() != topicName.getPartitionIndex()) {
- log.warn("[{}] Invalid parameter for expire message by position, partition index of passed in message"
- + " position {} doesn't match partition index of topic requested {}.",
- clientAppId(), messageId, topicName);
- asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
- "Invalid parameter for expire message by position, partition index of message position "
- + "passed in doesn't match partition index for the topic."));
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
- PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
- if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
- return;
- }
- try {
- PersistentSubscription sub = topic.getSubscription(subName);
- if (sub == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
- return;
- }
- CompletableFuture<Integer> batchSizeFuture = new CompletableFuture<>();
- getEntryBatchSize(batchSizeFuture, topic, messageId, batchIndex);
- batchSizeFuture.thenAccept(bi -> {
- PositionImpl position = calculatePositionAckSet(isExcluded, bi, batchIndex, messageId);
- boolean issued;
- try {
- if (subName.startsWith(topic.getReplicatorPrefix())) {
- String remoteCluster = PersistentReplicator.getRemoteCluster(subName);
- PersistentReplicator repl = (PersistentReplicator)
- topic.getPersistentReplicator(remoteCluster);
- checkNotNull(repl);
- issued = repl.expireMessages(position);
- } else {
- checkNotNull(sub);
- issued = sub.expireMessages(position);
- }
- if (issued) {
- log.info("[{}] Message expire started up to {} on {} {}", clientAppId(), position,
- topicName, subName);
- } else {
- if (log.isDebugEnabled()) {
- log.debug("Expire message by position not issued on topic {} for subscription {} "
- + "due to ongoing message expiration not finished or subscription "
- + "almost catch up.", topicName, subName);
- }
- throw new RestException(Status.CONFLICT, "Expire message by position not issued on topic "
- + topicName + " for subscription " + subName + " due to ongoing message expiration"
- + " not finished or invalid message position provided.");
- }
- } catch (NullPointerException npe) {
- throw new RestException(Status.NOT_FOUND, "Subscription not found");
- } catch (Exception exception) {
- log.error("[{}] Failed to expire messages up to {} on {} with subscription {} {}",
- clientAppId(), position, topicName, subName, exception);
- throw new RestException(exception);
- }
- }).exceptionally(e -> {
- log.error("[{}] Failed to expire messages up to {} on {} with subscription {} {}", clientAppId(),
- messageId, topicName, subName, e);
- asyncResponse.resume(e);
- return null;
- });
- } catch (Exception e) {
- log.warn("[{}][{}] Failed to expire messages up to {} on subscription {} to position {}",
- clientAppId(), topicName, messageId, subName, messageId, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- }
+ future = CompletableFuture.completedFuture(null);
}
- asyncResponse.resume(Response.noContent().build());
+ future.thenAccept(__ -> {
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(unused -> validateTopicOperationAsync(topicName, TopicOperation.EXPIRE_MESSAGES))
+ .thenAccept(unused -> {
+ log.info("[{}][{}] received expire messages on subscription {} to position {}", clientAppId(),
+ topicName, subName, messageId);
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(partitionMetadata -> {
+ if (!topicName.isPartitioned() && partitionMetadata.partitions > 0) {
+ log.warn("[{}] Not supported operation expire message up to {} on "
+ + "partitioned-topic {} {}", clientAppId(), messageId, topicName,
+ subName);
+ asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
+ "Expire message at position is not supported for partitioned-topic"));
+ return;
+ } else if (messageId.getPartitionIndex() != topicName.getPartitionIndex()) {
+ log.warn("[{}] Invalid parameter for expire message by position, partition "
+ + "index of passed in message position {} doesn't match partition "
+ + "index of topic requested {}.", clientAppId(), messageId, topicName);
+ asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid parameter for expire message by position, partition index of "
+ + "message position passed in doesn't match partition index for the "
+ + "topic."));
+ return;
+ } else {
+ getTopicReferenceAsync(topicName).thenAccept(t -> {
+ PersistentTopic topic = (PersistentTopic) t;
+ if (topic == null) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND,
+ "Topic not found"));
+ return;
+ }
+ try {
+ PersistentSubscription sub = topic.getSubscription(subName);
+ if (sub == null) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND,
+ "Subscription not found"));
+ return;
+ }
+ CompletableFuture<Integer> batchSizeFuture = new CompletableFuture<>();
+ getEntryBatchSize(batchSizeFuture, topic, messageId, batchIndex);
+ batchSizeFuture.thenAccept(bi -> {
+ PositionImpl position = calculatePositionAckSet(isExcluded, bi,
+ batchIndex, messageId);
+ boolean issued;
+ try {
+ if (subName.startsWith(topic.getReplicatorPrefix())) {
+ String remoteCluster = PersistentReplicator
+ .getRemoteCluster(subName);
+ PersistentReplicator repl = (PersistentReplicator)
+ topic.getPersistentReplicator(remoteCluster);
+ checkNotNull(repl);
+ issued = repl.expireMessages(position);
+ } else {
+ checkNotNull(sub);
+ issued = sub.expireMessages(position);
+ }
+ if (issued) {
+ log.info("[{}] Message expire started up to {} on {} {}",
+ clientAppId(), position, topicName, subName);
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("Expire message by position not issued on "
+ + "topic {} for subscription {} due to ongoing "
+ + "message expiration not finished or "
+ + "subscription almost catch up.", topicName,
+ subName);
+ }
+ throw new RestException(Status.CONFLICT, "Expire message by"
+ + " position not issued on topic " + topicName
+ + " for subscription " + subName + " due to ongoing"
+ + " message expiration not finished or invalid "
+ + "message position provided.");
+ }
+ } catch (NullPointerException npe) {
Review comment:
Same as https://github.com/apache/pulsar/pull/13876#discussion_r789430042, please help is it necessary for now?
--
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] liudezhi2098 commented on pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
Posted by GitBox <gi...@apache.org>.
liudezhi2098 commented on pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878#issuecomment-1022748657
/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] liudezhi2098 commented on pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
Posted by GitBox <gi...@apache.org>.
liudezhi2098 commented on pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878#issuecomment-1018287805
/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] hangc0276 merged pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
Posted by GitBox <gi...@apache.org>.
hangc0276 merged pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878
--
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] liudezhi2098 commented on pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
Posted by GitBox <gi...@apache.org>.
liudezhi2098 commented on pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878#issuecomment-1018495827
> emmm. i think we need to refactor this code. it seems very like callback hell.
>
> @liudezhi2098 Do you have any good idea ?
The code is not very nice，adjust the style。
--
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] liudezhi2098 commented on pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
Posted by GitBox <gi...@apache.org>.
liudezhi2098 commented on pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878#issuecomment-1022733776
/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] Technoboy- commented on a change in pull request #13878: [Broker] Fix call sync method in async rest api for internalExpireMessagesByPosition
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #13878:
URL: https://github.com/apache/pulsar/pull/13878#discussion_r792459420
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -3447,47 +3447,77 @@ private void internalExpireMessagesByTimestampForSinglePartition(String subName,
protected void internalExpireMessagesByPosition(AsyncResponse asyncResponse, String subName, boolean authoritative,
MessageIdImpl messageId, boolean isExcluded, int batchIndex) {
+ CompletableFuture<Void> future;
if (topicName.isGlobal()) {
- try {
- validateGlobalNamespaceOwnership(namespaceName);
- } catch (Exception e) {
- log.warn("[{}][{}] Failed to expire messages on subscription {} to position {}: {}", clientAppId(),
- topicName, subName, messageId, e.getMessage());
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ } else {
+ future = CompletableFuture.completedFuture(null);
}
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.EXPIRE_MESSAGES);
-
- log.info("[{}][{}] received expire messages on subscription {} to position {}", clientAppId(), topicName,
- subName, messageId);
+ future.thenAccept(__ -> {
+ validateTopicOperationAsync(topicName, TopicOperation.EXPIRE_MESSAGES)
+ .thenCompose(unused -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenAccept(unused -> {
Review comment:
I think it's better to use thenCompose here :
thenCompose(unused -> getPartitionedTopicMetadataAsync())
Then we can delete 3481~3485
--
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