You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2023/03/02 09:42:54 UTC
[GitHub] [pulsar] poorbarcode opened a new pull request, #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
poorbarcode opened a new pull request, #19692:
URL: https://github.com/apache/pulsar/pull/19692
### Motivation
After a Topic close failure or a delete failure, the fence state will be reset to get the topic back to work,but it will not reset the fence state of the subscription, which will result in the consumer never being able to connect to the broker.
### Modifications
Reset the fence state of subscriptions if the operation of topic close is failed.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository:
- https://github.com/poorbarcode/pulsar/pull/73
--
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] poorbarcode commented on pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#issuecomment-1458014859
/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] poorbarcode commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1122849269
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3230,6 +3230,7 @@ private void fenceTopicToCloseOrDelete() {
}
private void unfenceTopicToResume() {
+ subscriptions.values().forEach(sub -> sub.resumeAfterFence());
Review Comment:
During the execution of the method `resumeAfterFence`, consumers will keep trying to reconnect. When the execution of method A is complete, the consumer will automatically connect successfully and can consume the message. So it's okay not to wait.
--
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 #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1128013949
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,52 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
+ }
+ fenceFuture = new CompletableFuture<>();
// block any further consumers on this subscription
IS_FENCED_UPDATER.set(this, TRUE);
(dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null))
.thenCompose(v -> close()).thenRun(() -> {
log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName);
- disconnectFuture.complete(null);
+ fenceFuture.complete(null);
}).exceptionally(exception -> {
- IS_FENCED_UPDATER.set(this, FALSE);
- if (dispatcher != null) {
- dispatcher.reset();
- }
log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName,
exception);
- disconnectFuture.completeExceptionally(exception);
+ fenceFuture.completeExceptionally(exception);
+ resumeAfterFence();
return null;
});
+ return fenceFuture;
+ }
- return disconnectFuture;
+ /**
+ * Resume subscription after topic deletion or close failure.
+ */
+ public synchronized void resumeAfterFence() {
+ // If "fenceFuture" is null, it means that "disconnect" has never been called.
+ if (fenceFuture != null) {
+ fenceFuture.whenComplete((ignore, ignoreEx) -> {
+ synchronized (PersistentSubscription.this) {
+ try {
+ if (IS_FENCED_UPDATER.compareAndSet(this, TRUE, FALSE)) {
+ if (dispatcher != null) {
+ dispatcher.reset();
+ }
+ }
+ fenceFuture = null;
+ } catch (Exception ex) {
+ log.error("[{}] Resume subscription[{}] failure: {}", topicName, subName, ex.getMessage(), ex);
Review Comment:
```suggestion
log.error("[{}] Resume subscription [{}] failure", topicName, subName, ex);
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,52 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
Review Comment:
It looks like we can simplify to
```java
if (fenceFuture != null && !fenceFuture.isCompletedExceptionally()) {
return fenceFuture;
}
```
And why must we create a new future if the existing `fenceFuture` is completed with an exception? To make it retryable?
If yes, we'd better add a comment here to help us to understand why it should be retryable in the future.
--
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] gaoran10 commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1123997298
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,62 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
Review Comment:
It seems that if PersistentSubscription disconnect successfully, but PersistentTopic close failed and resume, the PersistentSubscription can't disconnect again.
--
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] poorbarcode commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1125815065
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,62 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
Review Comment:
Good catch. already 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] poorbarcode merged pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode merged PR #19692:
URL: https://github.com/apache/pulsar/pull/19692
--
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] poorbarcode commented on pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#issuecomment-1460150462
/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] eolivelli commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "eolivelli (via GitHub)" <gi...@apache.org>.
eolivelli commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1126063515
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3230,6 +3230,7 @@ private void fenceTopicToCloseOrDelete() {
}
private void unfenceTopicToResume() {
+ subscriptions.values().forEach(sub -> sub.resumeAfterFence());
Review Comment:
Then you can make resumeAfterFence return "void", can't you ?
--
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] poorbarcode commented on pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#issuecomment-1459884581
/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] poorbarcode commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1126070013
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3230,6 +3230,7 @@ private void fenceTopicToCloseOrDelete() {
}
private void unfenceTopicToResume() {
+ subscriptions.values().forEach(sub -> sub.resumeAfterFence());
Review Comment:
Good suggestion. Already 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] eolivelli commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "eolivelli (via GitHub)" <gi...@apache.org>.
eolivelli commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1122835105
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,54 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
+ }
+ fenceFuture = new CompletableFuture<>();
// block any further consumers on this subscription
IS_FENCED_UPDATER.set(this, TRUE);
(dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null))
.thenCompose(v -> close()).thenRun(() -> {
log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName);
- disconnectFuture.complete(null);
+ fenceFuture.complete(null);
}).exceptionally(exception -> {
IS_FENCED_UPDATER.set(this, FALSE);
if (dispatcher != null) {
dispatcher.reset();
}
log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName,
exception);
- disconnectFuture.completeExceptionally(exception);
+ fenceFuture.completeExceptionally(exception);
return null;
});
- return disconnectFuture;
+ return fenceFuture;
+ }
+
+ /**
+ * Resume subscription after topic deletion or close failure.
+ */
+ public synchronized CompletableFuture<Void> resumeAfterFence() {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ fenceFuture.whenComplete((ignore, ignore2) -> {
Review Comment:
fenceFuture may be null here
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,54 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
+ }
+ fenceFuture = new CompletableFuture<>();
// block any further consumers on this subscription
IS_FENCED_UPDATER.set(this, TRUE);
(dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null))
.thenCompose(v -> close()).thenRun(() -> {
log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName);
- disconnectFuture.complete(null);
+ fenceFuture.complete(null);
}).exceptionally(exception -> {
IS_FENCED_UPDATER.set(this, FALSE);
if (dispatcher != null) {
dispatcher.reset();
}
log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName,
exception);
- disconnectFuture.completeExceptionally(exception);
+ fenceFuture.completeExceptionally(exception);
return null;
});
- return disconnectFuture;
+ return fenceFuture;
+ }
+
+ /**
+ * Resume subscription after topic deletion or close failure.
+ */
+ public synchronized CompletableFuture<Void> resumeAfterFence() {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ fenceFuture.whenComplete((ignore, ignore2) -> {
+ try {
+ if (IS_FENCED_UPDATER.compareAndSet(this, TRUE, FALSE)) {
+ if (dispatcher != null) {
+ dispatcher.reset();
+ }
+ }
+ result.complete(null);
+ } catch (Exception ex){
+ result.completeExceptionally(ex);
+ log.error("[{}] Resume subscription[{}] failure: {}", topicName, subName, ex.getMessage(), ex);
Review Comment:
please move this before completeExceptionally, otherwise logs may be hard to understand
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3230,6 +3230,7 @@ private void fenceTopicToCloseOrDelete() {
}
private void unfenceTopicToResume() {
+ subscriptions.values().forEach(sub -> sub.resumeAfterFence());
Review Comment:
resumeAfterFence returns a CompletableFuture, don't we need to "wait"
--
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] poorbarcode commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1128043073
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,52 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
+ }
+ fenceFuture = new CompletableFuture<>();
// block any further consumers on this subscription
IS_FENCED_UPDATER.set(this, TRUE);
(dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null))
.thenCompose(v -> close()).thenRun(() -> {
log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName);
- disconnectFuture.complete(null);
+ fenceFuture.complete(null);
}).exceptionally(exception -> {
- IS_FENCED_UPDATER.set(this, FALSE);
- if (dispatcher != null) {
- dispatcher.reset();
- }
log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName,
exception);
- disconnectFuture.completeExceptionally(exception);
+ fenceFuture.completeExceptionally(exception);
+ resumeAfterFence();
return null;
});
+ return fenceFuture;
+ }
- return disconnectFuture;
+ /**
+ * Resume subscription after topic deletion or close failure.
+ */
+ public synchronized void resumeAfterFence() {
+ // If "fenceFuture" is null, it means that "disconnect" has never been called.
+ if (fenceFuture != null) {
+ fenceFuture.whenComplete((ignore, ignoreEx) -> {
+ synchronized (PersistentSubscription.this) {
+ try {
+ if (IS_FENCED_UPDATER.compareAndSet(this, TRUE, FALSE)) {
+ if (dispatcher != null) {
+ dispatcher.reset();
+ }
+ }
+ fenceFuture = null;
+ } catch (Exception ex) {
+ log.error("[{}] Resume subscription[{}] failure: {}", topicName, subName, ex.getMessage(), ex);
Review Comment:
already changed
--
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] poorbarcode commented on pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#issuecomment-1461155886
/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] poorbarcode commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1122846325
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,54 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
+ }
+ fenceFuture = new CompletableFuture<>();
// block any further consumers on this subscription
IS_FENCED_UPDATER.set(this, TRUE);
(dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null))
.thenCompose(v -> close()).thenRun(() -> {
log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName);
- disconnectFuture.complete(null);
+ fenceFuture.complete(null);
}).exceptionally(exception -> {
IS_FENCED_UPDATER.set(this, FALSE);
if (dispatcher != null) {
dispatcher.reset();
}
log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName,
exception);
- disconnectFuture.completeExceptionally(exception);
+ fenceFuture.completeExceptionally(exception);
return null;
});
- return disconnectFuture;
+ return fenceFuture;
+ }
+
+ /**
+ * Resume subscription after topic deletion or close failure.
+ */
+ public synchronized CompletableFuture<Void> resumeAfterFence() {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ fenceFuture.whenComplete((ignore, ignore2) -> {
Review Comment:
Already fixed
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,54 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
+ }
+ fenceFuture = new CompletableFuture<>();
// block any further consumers on this subscription
IS_FENCED_UPDATER.set(this, TRUE);
(dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null))
.thenCompose(v -> close()).thenRun(() -> {
log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName);
- disconnectFuture.complete(null);
+ fenceFuture.complete(null);
}).exceptionally(exception -> {
IS_FENCED_UPDATER.set(this, FALSE);
if (dispatcher != null) {
dispatcher.reset();
}
log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName,
exception);
- disconnectFuture.completeExceptionally(exception);
+ fenceFuture.completeExceptionally(exception);
return null;
});
- return disconnectFuture;
+ return fenceFuture;
+ }
+
+ /**
+ * Resume subscription after topic deletion or close failure.
+ */
+ public synchronized CompletableFuture<Void> resumeAfterFence() {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ fenceFuture.whenComplete((ignore, ignore2) -> {
+ try {
+ if (IS_FENCED_UPDATER.compareAndSet(this, TRUE, FALSE)) {
+ if (dispatcher != null) {
+ dispatcher.reset();
+ }
+ }
+ result.complete(null);
+ } catch (Exception ex){
+ result.completeExceptionally(ex);
+ log.error("[{}] Resume subscription[{}] failure: {}", topicName, subName, ex.getMessage(), ex);
Review Comment:
Already 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] poorbarcode commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1122849269
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -3230,6 +3230,7 @@ private void fenceTopicToCloseOrDelete() {
}
private void unfenceTopicToResume() {
+ subscriptions.values().forEach(sub -> sub.resumeAfterFence());
Review Comment:
During the execution of the method `resumeAfterFence`, consumers will keep trying to reconnect. When the execution of method `resumeAfterFence` is complete, the consumer will automatically connect successfully and can consume the message. So it's okay not to wait.
--
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] poorbarcode commented on pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#issuecomment-1458288459
/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] poorbarcode commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1128043437
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,52 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
Review Comment:
I think you are right. There has no scenario that needs to retry.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,52 @@ public CompletableFuture<Void> close() {
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null && !fenceFuture.isDone()){
+ return fenceFuture;
+ }
+ if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+ return fenceFuture;
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