heesung-sn commented on code in PR #21682:
URL: https://github.com/apache/pulsar/pull/21682#discussion_r1421046533
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -889,53 +890,79 @@ public int getTotalNonContiguousDeletedMessagesRange() {
*
* @return CompletableFuture indicating the completion of delete operation
*/
- @Override
- public CompletableFuture<Void> close() {
- synchronized (this) {
- if (dispatcher != null && dispatcher.isConsumerConnected()) {
- return FutureUtil.failedFuture(new
SubscriptionBusyException("Subscription has active consumers"));
- }
- return this.pendingAckHandle.closeAsync().thenAccept(v -> {
- IS_FENCED_UPDATER.set(this, TRUE);
- log.info("[{}][{}] Successfully closed subscription [{}]",
topicName, subName, cursor);
- });
+ private synchronized CompletableFuture<Void> close(boolean
checkActiveConsumers) {
+ if (checkActiveConsumers && dispatcher != null &&
dispatcher.isConsumerConnected()) {
+ return FutureUtil.failedFuture(new
SubscriptionBusyException("Subscription has active consumers"));
}
+ return this.pendingAckHandle.closeAsync().thenAccept(v -> {
+ IS_FENCED_UPDATER.set(this, TRUE);
+ log.info("[{}][{}] Successfully closed subscription [{}]",
topicName, subName, cursor);
+ });
+ }
+
+
+ /**
+ * Disconnect all consumers from this subscription.
+ *
+ * @return CompletableFuture indicating the completion of the operation.
+ */
+ @Override
+ public synchronized CompletableFuture<Void>
disconnect(Optional<BrokerLookupData> assignedBrokerLookupData) {
+ CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+
+ (dispatcher != null
+ ? dispatcher.disconnectAllConsumers(false,
assignedBrokerLookupData)
+ : CompletableFuture.completedFuture(null))
+ .thenRun(() -> {
+ log.info("[{}][{}] Successfully disconnected subscription
consumers", topicName, subName);
+ disconnectFuture.complete(null);
+ }).exceptionally(exception -> {
+ log.error("[{}][{}] Error disconnecting subscription
consumers", topicName, subName, exception);
+ disconnectFuture.completeExceptionally(exception);
+ return null;
+ });
+
+ return disconnectFuture;
}
/**
- * Disconnect all consumers attached to the dispatcher and close this
subscription.
+ * Fence this subscription and optionally disconnect all consumers.
*
- * @return CompletableFuture indicating the completion of disconnect
operation
+ * @return CompletableFuture indicating the completion of the operation.
*/
@Override
- public synchronized CompletableFuture<Void> disconnect() {
- if (fenceFuture != null){
+ public synchronized CompletableFuture<Void> close(boolean
disconnectConsumers,
+
Optional<BrokerLookupData> assignedBrokerLookupData) {
+ if (fenceFuture != null) {
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);
+ (dispatcher != null
+ ? dispatcher.close(disconnectConsumers,
assignedBrokerLookupData)
+ : CompletableFuture.completedFuture(null))
+ .thenCompose(v -> close(false)).thenRun(() -> {
Review Comment:
ok. plz add a comment with this reason.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -308,7 +309,7 @@ public synchronized void removeConsumer(Consumer consumer,
boolean isResetCursor
if (!cursor.isDurable()) {
// If cursor is not durable, we need to clean up the
subscription as well
- this.close().thenRun(() -> {
+ this.close(false).thenRun(() -> {
Review Comment:
ok. plz add a comment with this reason.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
