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/03/26 05:40:42 UTC
[GitHub] [pulsar] mattisonchao opened a new pull request #14887: [feature][admin] Support reset cluster replicator cursor
mattisonchao opened a new pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887
### Motivation
Currently, we do not support resetting the cursor position on the cluster replicator.
After the check, I think we could support it based on ``PersistentTopics#resetCursorOnPosition``.
#### Usage
**Admin**
```java
admin1.topics().resetCursor(topicName.toString(), "pulsar.repl.r2", new MessageIdImpl(0,0,-1));
```
**cli**
```bash
bin/pulsar-admin topics reset-cursor -s pulsar.repl.r2 -m 0:0:-1 persistent://xxxx/xx/xxx
```
### Modifications
- Support for resetting cluster replicator cursor based on ``PersistentTopics#resetCursorOnPosition`` method
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
- The admin cli options: (yes)
### Documentation
- [x] `doc-required`
--
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 a change in pull request #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#discussion_r835910425
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
##########
@@ -224,6 +224,10 @@ public static String getReplicatorName(String replicatorPrefix, String cluster)
return (replicatorPrefix + "." + cluster).intern();
}
+ public static String getClusterName(String replicatorPrefix, String replicatorName) {
+ return replicatorName.replace(replicatorPrefix + ".", "");
Review comment:
@eolivelli
fixed. PTAL :)
--
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 a change in pull request #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#discussion_r839151588
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
##########
@@ -771,10 +784,55 @@ public boolean isConnected() {
return producer != null && producer.isConnected();
}
+ @Override
+ public CompletableFuture<Void> resetCursor(Position position) {
+ CURSOR_RESETTING_UPDATER.set(this, TRUE);
Review comment:
the ``cursor.asyncResetCursor`` will ensure that.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -2278,89 +2279,60 @@ private void internalCreateSubscriptionForNonPartitionedTopic(
});
}
- protected void internalResetCursorOnPosition(AsyncResponse asyncResponse, String subName, boolean authoritative,
+ protected CompletableFuture<Void> internalResetCursorOnPosition(String name, boolean authoritative,
Review comment:
because if you want to reset the replicator cursor that would be a replicator.
I'm not sure ``subName``is ambiguous.
--
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 #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#issuecomment-1079802852
/pulsarbot rerun-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] RobertIndie commented on a change in pull request #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
RobertIndie commented on a change in pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#discussion_r837004332
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -2278,89 +2279,60 @@ private void internalCreateSubscriptionForNonPartitionedTopic(
});
}
- protected void internalResetCursorOnPosition(AsyncResponse asyncResponse, String subName, boolean authoritative,
+ protected CompletableFuture<Void> internalResetCursorOnPosition(String name, boolean authoritative,
Review comment:
Why change `subName` to `name` here?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
##########
@@ -771,10 +784,55 @@ public boolean isConnected() {
return producer != null && producer.isConnected();
}
+ @Override
+ public CompletableFuture<Void> resetCursor(Position position) {
+ CURSOR_RESETTING_UPDATER.set(this, TRUE);
Review comment:
Is there a race condition here when multiple threads call this method concurrently?
--
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] momo-jun commented on pull request #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
momo-jun commented on pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#issuecomment-1080302900
#### Doc status update
@mattisonchao will contribute a doc PR as soon as this PR is merged.
Scoping: Describe the enhanced reset method in the **Admin API > Topics > Manage topic resources > Reset cursor** section.
--
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] eolivelli commented on a change in pull request #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#discussion_r835900784
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
##########
@@ -224,6 +224,10 @@ public static String getReplicatorName(String replicatorPrefix, String cluster)
return (replicatorPrefix + "." + cluster).intern();
}
+ public static String getClusterName(String replicatorPrefix, String replicatorName) {
+ return replicatorName.replace(replicatorPrefix + ".", "");
Review comment:
Thisbis not valid.
We can strip out this string only if it is in the beginning
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
##########
@@ -622,10 +624,30 @@ public void resetCursorOnPosition(@Suspended final AsyncResponse asyncResponse,
boolean authoritative, ResetCursorData resetCursorData) {
try {
validateTopicName(property, cluster, namespace, encodedTopic);
- internalResetCursorOnPosition(asyncResponse, decode(encodedSubName), authoritative,
- new MessageIdImpl(resetCursorData.getLedgerId(),
- resetCursorData.getEntryId(), resetCursorData.getPartitionIndex())
- , resetCursorData.isExcluded(), resetCursorData.getBatchIndex());
+ String decodeSubName = decode(encodedSubName);
+ MessageIdImpl messageId = new MessageIdImpl(resetCursorData.getLedgerId(), resetCursorData.getEntryId(),
+ resetCursorData.getPartitionIndex());
+ internalResetCursorOnPosition(decode(encodedSubName), authoritative, messageId,
Review comment:
We already computed 'decode'
--
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 a change in pull request #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#discussion_r835910409
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
##########
@@ -622,10 +624,30 @@ public void resetCursorOnPosition(@Suspended final AsyncResponse asyncResponse,
boolean authoritative, ResetCursorData resetCursorData) {
try {
validateTopicName(property, cluster, namespace, encodedTopic);
- internalResetCursorOnPosition(asyncResponse, decode(encodedSubName), authoritative,
- new MessageIdImpl(resetCursorData.getLedgerId(),
- resetCursorData.getEntryId(), resetCursorData.getPartitionIndex())
- , resetCursorData.isExcluded(), resetCursorData.getBatchIndex());
+ String decodeSubName = decode(encodedSubName);
+ MessageIdImpl messageId = new MessageIdImpl(resetCursorData.getLedgerId(), resetCursorData.getEntryId(),
+ resetCursorData.getPartitionIndex());
+ internalResetCursorOnPosition(decode(encodedSubName), authoritative, messageId,
Review comment:
fixed.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
##########
@@ -224,6 +224,10 @@ public static String getReplicatorName(String replicatorPrefix, String cluster)
return (replicatorPrefix + "." + cluster).intern();
}
+ public static String getClusterName(String replicatorPrefix, String replicatorName) {
+ return replicatorName.replace(replicatorPrefix + ".", "");
Review comment:
fixed.
--
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] Anonymitaet commented on pull request #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
Anonymitaet commented on pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#issuecomment-1080088011
@momo-jun a soft reminder: here is a PR w/ doc-required label, could u pls follow up? Thanks
--
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] RobertIndie commented on a change in pull request #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
RobertIndie commented on a change in pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#discussion_r839242885
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -2278,89 +2279,60 @@ private void internalCreateSubscriptionForNonPartitionedTopic(
});
}
- protected void internalResetCursorOnPosition(AsyncResponse asyncResponse, String subName, boolean authoritative,
+ protected CompletableFuture<Void> internalResetCursorOnPosition(String name, boolean authoritative,
Review comment:
But the `name` doesn't make codes more readable. The replicator is also the subscription even if it's a special subscription. I think it's better to reserve `subName` if we can't find a better name for it. How about using a more specific name like `cursorName`?
--
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] RobertIndie commented on a change in pull request #14887: [feature][admin] Support reset cluster replicator cursor
Posted by GitBox <gi...@apache.org>.
RobertIndie commented on a change in pull request #14887:
URL: https://github.com/apache/pulsar/pull/14887#discussion_r839246936
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
##########
@@ -771,10 +784,55 @@ public boolean isConnected() {
return producer != null && producer.isConnected();
}
+ @Override
+ public CompletableFuture<Void> resetCursor(Position position) {
+ CURSOR_RESETTING_UPDATER.set(this, TRUE);
Review comment:
But there may be multiple threads calling `cursor.asyncResetCursor` concurrently as well. I think we need to use `compareAndSet` here. Please correct me if I have missed something.
--
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