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 2018/07/13 13:55:35 UTC
[GitHub] merlimat closed pull request #2149: Avoid closing cursor when
consumers are disconnected
merlimat closed pull request #2149: Avoid closing cursor when consumers are disconnected
URL: https://github.com/apache/incubator-pulsar/pull/2149
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index 54ac65f4bf..8a77bfacb5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -54,6 +54,7 @@
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ConsumerStats;
import org.apache.pulsar.common.policies.data.SubscriptionStats;
+import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.utils.CopyOnWriteArrayList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -494,36 +495,15 @@ public int getTotalNonContiguousDeletedMessagesRange() {
*/
@Override
public CompletableFuture<Void> close() {
- CompletableFuture<Void> closeFuture = new CompletableFuture<>();
-
synchronized (this) {
if (dispatcher != null && dispatcher.isConsumerConnected()) {
- closeFuture.completeExceptionally(new SubscriptionBusyException("Subscription has active consumers"));
- return closeFuture;
+ return FutureUtil.failedFuture(new SubscriptionBusyException("Subscription has active consumers"));
}
IS_FENCED_UPDATER.set(this, TRUE);
- log.info("[{}][{}] Successfully fenced cursor ledger [{}]", topicName, subName, cursor);
+ log.info("[{}][{}] Successfully closed subscription [{}]", topicName, subName, cursor);
}
- cursor.asyncClose(new CloseCallback() {
- @Override
- public void closeComplete(Object ctx) {
- if (log.isDebugEnabled()) {
- log.debug("[{}][{}] Successfully closed cursor ledger", topicName, subName);
- }
- closeFuture.complete(null);
- }
-
- @Override
- public void closeFailed(ManagedLedgerException exception, Object ctx) {
- IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
-
- log.error("[{}][{}] Error closing cursor for subscription", topicName, subName, exception);
- closeFuture.completeExceptionally(new PersistenceException(exception));
- }
- }, null);
-
- return closeFuture;
+ return CompletableFuture.completedFuture(null);
}
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 7db3e71181..c66e10bbc3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -563,9 +563,13 @@ public void openCursorComplete(ManagedCursor cursor, Object ctx) {
@Override
public void openCursorFailed(ManagedLedgerException exception, Object ctx) {
- log.warn("[{}] Failed to create subscription for {}", topic, subscriptionName);
+ log.warn("[{}] Failed to create subscription for {}: {}", topic, subscriptionName, exception.getMessage());
USAGE_COUNT_UPDATER.decrementAndGet(PersistentTopic.this);
subscriptionFuture.completeExceptionally(new PersistenceException(exception));
+ if (exception instanceof ManagedLedgerFencedException) {
+ // If the managed ledger has been fenced, we cannot continue using it. We need to close and reopen
+ close();
+ }
}
}, null);
return subscriptionFuture;
@@ -675,14 +679,14 @@ void removeSubscription(String subscriptionName) {
* Forcefully close all producers/consumers/replicators and deletes the topic. this function is used when local
* cluster is removed from global-namespace replication list. Because broker doesn't allow lookup if local cluster
* is not part of replication cluster list.
- *
+ *
* @return
*/
@Override
public CompletableFuture<Void> deleteForcefully() {
return delete(false, true);
}
-
+
/**
* Delete the managed ledger associated with this topic
*
@@ -693,7 +697,7 @@ void removeSubscription(String subscriptionName) {
* @param closeIfClientsConnected
* Flag indicate whether explicitly close connected producers/consumers/replicators before trying to delete topic. If
* any client is connected to a topic and if this flag is disable then this operation fails.
- *
+ *
* @return Completable future indicating completion of delete operation Completed exceptionally with:
* IllegalStateException if topic is still active ManagedLedgerException if ledger delete operation fails
*/
@@ -707,7 +711,7 @@ void removeSubscription(String subscriptionName) {
deleteFuture.completeExceptionally(new TopicFencedException("Topic is already fenced"));
return deleteFuture;
}
-
+
CompletableFuture<Void> closeClientFuture = new CompletableFuture<>();
if (closeIfClientsConnected) {
List<CompletableFuture<Void>> futures = Lists.newArrayList();
@@ -908,7 +912,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
}
String localCluster = brokerService.pulsar().getConfiguration().getClusterName();
-
+
// if local cluster is removed from global namespace cluster-list : then delete topic forcefully because pulsar
// doesn't serve global topic without local repl-cluster configured.
if (TopicName.get(topic).isGlobal() && !configuredClusters.contains(localCluster)) {
@@ -940,7 +944,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
futures.add(removeReplicator(cluster));
}
}
-
+
});
return FutureUtil.waitForAll(futures);
@@ -1057,7 +1061,7 @@ protected boolean addReplicationCluster(String remoteCluster, PersistentTopic pe
CompletableFuture<Void> removeReplicator(String remoteCluster) {
log.info("[{}] Removing replicator to {}", topic, remoteCluster);
final CompletableFuture<Void> future = new CompletableFuture<>();
-
+
String name = PersistentReplicator.getReplicatorName(replicatorPrefix, remoteCluster);
replicators.get(remoteCluster).disconnect().thenRun(() -> {
@@ -1179,7 +1183,7 @@ public void updateRates(NamespaceStats nsStats, NamespaceBundleStats bundleStats
} catch (Exception e) {
log.warn("[{}] Failed to update cursro state ", topic, e);
}
-
+
// Update replicator stats
ReplicatorStats rStat = replicator.getStats();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java
index 60e7cd680a..895d78862a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java
@@ -39,7 +39,7 @@
/**
* Test that sequence id from a producer is correct when there are send errors
*/
- @Test(enabled = false)
+ @Test
public void testCheckSequenceId() throws Exception {
admin.namespaces().createNamespace("prop/my-test", Collections.singleton("usc"));
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services