This is an automated email from the ASF dual-hosted git repository. mmerli pushed a commit to branch branch-3.0 in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push: new b0542b3312e [improve] Retry re-validating ResourceLock with backoff after errors (#22617) b0542b3312e is described below commit b0542b3312ef915e7467d9550d19a3e7ca6aca99 Author: Matteo Merli <mme...@apache.org> AuthorDate: Tue May 7 10:35:23 2024 -0700 [improve] Retry re-validating ResourceLock with backoff after errors (#22617) --- .../pulsar/broker/service/AbstractReplicator.java | 2 +- .../service/PulsarMetadataEventSynchronizer.java | 2 +- .../broker/service/TopicPoliciesService.java | 4 +-- .../PersistentDispatcherMultipleConsumers.java | 2 +- .../PersistentDispatcherSingleActiveConsumer.java | 2 +- .../service/persistent/PersistentReplicator.java | 2 +- .../streamingdispatch/StreamingEntryReader.java | 2 +- .../pendingack/impl/PendingAckHandleImpl.java | 2 +- .../common/naming/NamespaceBundleFactory.java | 2 +- .../SystemTopicBasedTopicPoliciesServiceTest.java | 4 +-- .../pulsar/client/impl/ConnectionHandlerTest.java | 2 ++ .../apache/pulsar/client/impl/RetryUtilTest.java | 2 ++ .../client/impl/BinaryProtoLookupService.java | 2 ++ .../pulsar/client/impl/ConnectionHandler.java | 1 + .../apache/pulsar/client/impl/ConsumerImpl.java | 2 ++ .../impl/PatternMultiTopicsConsumerImpl.java | 2 ++ .../apache/pulsar/client/impl/ProducerImpl.java | 1 + .../pulsar/client/impl/PulsarClientImpl.java | 2 ++ .../pulsar/client/impl/TopicListWatcher.java | 1 + .../client/impl/TransactionMetaStoreHandler.java | 2 ++ .../org/apache/pulsar/client/util/RetryUtil.java | 2 +- .../pulsar/client/impl/ConsumerImplTest.java | 1 + .../org/apache/pulsar/common/util}/Backoff.java | 2 +- .../apache/pulsar/common/util}/BackoffBuilder.java | 5 ++- .../apache/pulsar/common/util}/BackoffTest.java | 2 +- .../coordination/impl/LockManagerImpl.java | 10 +++--- .../coordination/impl/ResourceLockImpl.java | 37 +++++++++++++++++++--- .../apache/pulsar/metadata/LockManagerTest.java | 31 ++++++++++++++++++ 28 files changed, 106 insertions(+), 25 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java index 902420e77b9..e9911a3c5be 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java @@ -36,10 +36,10 @@ import org.apache.pulsar.client.api.MessageRoutingMode; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerBuilder; import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.client.impl.ProducerImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.util.Backoff; import org.apache.pulsar.common.util.FutureUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarMetadataEventSynchronizer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarMetadataEventSynchronizer.java index 80743e44ab7..0383a0b7552 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarMetadataEventSynchronizer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarMetadataEventSynchronizer.java @@ -33,8 +33,8 @@ import org.apache.pulsar.client.api.MessageRoutingMode; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; -import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.client.impl.PulsarClientImpl; +import org.apache.pulsar.common.util.Backoff; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.metadata.api.MetadataEvent; import org.apache.pulsar.metadata.api.MetadataEventSynchronizer; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java index c09bab0a4b6..cfac6b396e8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java @@ -24,13 +24,13 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import javax.annotation.Nonnull; import org.apache.pulsar.broker.service.BrokerServiceException.TopicPoliciesCacheNotInitException; -import org.apache.pulsar.client.impl.Backoff; -import org.apache.pulsar.client.impl.BackoffBuilder; import org.apache.pulsar.client.util.RetryUtil; import org.apache.pulsar.common.classification.InterfaceStability; import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.TopicPolicies; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.FutureUtil; /** diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java index 9d0dba798ad..2f99d57aa45 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java @@ -68,11 +68,11 @@ import org.apache.pulsar.broker.service.StickyKeyConsumerSelector; import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type; import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException; -import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; import org.apache.pulsar.common.api.proto.MessageMetadata; import org.apache.pulsar.common.policies.data.stats.TopicMetricBean; import org.apache.pulsar.common.protocol.Commands; +import org.apache.pulsar.common.util.Backoff; import org.apache.pulsar.common.util.Codec; import org.apache.pulsar.common.util.FutureUtil; import org.slf4j.Logger; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java index cc7b6841e5c..2830e700a62 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java @@ -50,8 +50,8 @@ import org.apache.pulsar.broker.service.SendMessageInfo; import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type; import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException; -import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; +import org.apache.pulsar.common.util.Backoff; import org.apache.pulsar.common.util.Codec; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java index 7ee1619e704..f5d0cc8aa9c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java @@ -55,7 +55,6 @@ import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClientException; -import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.client.impl.ProducerImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; @@ -64,6 +63,7 @@ import org.apache.pulsar.common.api.proto.MarkerType; import org.apache.pulsar.common.policies.data.stats.ReplicatorStatsImpl; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.stats.Rate; +import org.apache.pulsar.common.util.Backoff; import org.apache.pulsar.common.util.Codec; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java index 6ffc5ba0f62..bed3bea236f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java @@ -37,7 +37,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException; -import org.apache.pulsar.client.impl.Backoff; +import org.apache.pulsar.common.util.Backoff; /** * Entry reader that fulfill read request by streamline the read instead of reading with micro batch. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java index 5ed271c6fd4..9d07af4d26c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java @@ -59,11 +59,11 @@ import org.apache.pulsar.broker.transaction.pendingack.PendingAckStore; import org.apache.pulsar.broker.transaction.pendingack.TransactionPendingAckStoreProvider; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.transaction.TxnID; -import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.common.api.proto.CommandAck.AckType; import org.apache.pulsar.common.policies.data.TransactionInPendingAckStats; import org.apache.pulsar.common.policies.data.TransactionPendingAckStats; import org.apache.pulsar.common.stats.PositionInPendingAckStats; +import org.apache.pulsar.common.util.Backoff; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.RecoverTimeRecord; import org.apache.pulsar.common.util.collections.BitSetRecyclable; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java index 937d2763767..6f085cd5e2f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java @@ -50,10 +50,10 @@ import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerWrapper; import org.apache.pulsar.broker.resources.LocalPoliciesResources; import org.apache.pulsar.broker.resources.PulsarResources; -import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.common.policies.data.BundlesData; import org.apache.pulsar.common.policies.data.LocalPolicies; import org.apache.pulsar.common.policies.data.Policies; +import org.apache.pulsar.common.util.Backoff; import org.apache.pulsar.metadata.api.Notification; import org.apache.pulsar.policies.data.loadbalancer.BundleData; import org.apache.pulsar.stats.CacheMetricsCollector; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java index e571da13435..f79c7412bc7 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java @@ -45,8 +45,8 @@ import org.apache.pulsar.broker.service.BrokerServiceException.TopicPoliciesCach import org.apache.pulsar.broker.systopic.SystemTopicClient; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.impl.Backoff; -import org.apache.pulsar.client.impl.BackoffBuilder; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.events.PulsarEvent; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.TopicName; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionHandlerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionHandlerTest.java index f29d62db5f4..add0a5f3eef 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionHandlerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionHandlerTest.java @@ -32,6 +32,8 @@ import java.util.concurrent.atomic.AtomicInteger; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.FutureUtil; import org.awaitility.Awaitility; import org.awaitility.core.ConditionTimeoutException; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RetryUtilTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RetryUtilTest.java index 604c468b1de..58d1709cd1a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RetryUtilTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RetryUtilTest.java @@ -19,6 +19,8 @@ package org.apache.pulsar.client.impl; import org.apache.pulsar.client.util.RetryUtil; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.FutureUtil; import org.testng.annotations.Test; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java index 8ceb8e44975..9634c60f09a 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java @@ -45,6 +45,8 @@ import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.protocol.schema.BytesSchemaVersion; import org.apache.pulsar.common.schema.SchemaInfo; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.FutureUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java index 77a85c255a8..6403d48d7be 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java @@ -29,6 +29,7 @@ import lombok.Setter; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.impl.HandlerState.State; import org.apache.pulsar.common.protocol.Commands; +import org.apache.pulsar.common.util.Backoff; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java index 468b5f7a8b8..bebc5601c6a 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java @@ -117,6 +117,8 @@ import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.CompletableFutureCancellationHandler; import org.apache.pulsar.common.util.ExceptionHandler; import org.apache.pulsar.common.util.FutureUtil; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index 4d179f7d914..ec7ff7930c0 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -43,6 +43,8 @@ import org.apache.pulsar.common.lookup.GetTopicsResult; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.topics.TopicList; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.FutureUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java index a41de6d1051..426ac5009b5 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java @@ -91,6 +91,7 @@ import org.apache.pulsar.common.protocol.schema.SchemaHash; import org.apache.pulsar.common.protocol.schema.SchemaVersion; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.DateFormatter; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.RelativeTimeUtil; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index fdabb5fa8cf..b92e039e5fa 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -84,6 +84,8 @@ import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; import org.apache.pulsar.common.topics.TopicList; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.netty.EventLoopUtil; import org.slf4j.Logger; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java index 86adf69f06e..4e635e0d2e8 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java @@ -31,6 +31,7 @@ import org.apache.pulsar.common.api.proto.BaseCommand; import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.protocol.Commands; +import org.apache.pulsar.common.util.BackoffBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java index ebbfca0c3cb..c27fe958ddd 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java @@ -46,6 +46,8 @@ import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.api.proto.Subscription; import org.apache.pulsar.common.api.proto.TxnAction; import org.apache.pulsar.common.protocol.Commands; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java b/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java index 93501d7b6c1..912cb7d7c58 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java @@ -22,7 +22,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; -import org.apache.pulsar.client.impl.Backoff; +import org.apache.pulsar.common.util.Backoff; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java index 9995246c175..0c47d17098e 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java @@ -49,6 +49,7 @@ import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; import org.apache.pulsar.client.impl.conf.TopicConsumerConfigurationData; import org.apache.pulsar.client.util.ExecutorProvider; import org.apache.pulsar.client.util.ScheduledExecutorProvider; +import org.apache.pulsar.common.util.Backoff; import org.awaitility.Awaitility; import org.testng.Assert; import org.testng.annotations.AfterMethod; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/Backoff.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java similarity index 99% rename from pulsar-client/src/main/java/org/apache/pulsar/client/impl/Backoff.java rename to pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java index daaf3499400..4eab85f3c41 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/Backoff.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.pulsar.client.impl; +package org.apache.pulsar.common.util; import com.google.common.annotations.VisibleForTesting; import java.time.Clock; diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BackoffBuilder.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/BackoffBuilder.java similarity index 91% rename from pulsar-client/src/main/java/org/apache/pulsar/client/impl/BackoffBuilder.java rename to pulsar-common/src/main/java/org/apache/pulsar/common/util/BackoffBuilder.java index 9913393fa9a..69b39030081 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BackoffBuilder.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/BackoffBuilder.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.pulsar.client.impl; +package org.apache.pulsar.common.util; import java.time.Clock; import java.util.concurrent.TimeUnit; @@ -32,8 +32,11 @@ public class BackoffBuilder { public BackoffBuilder() { this.initial = 0; + this.unitInitial = TimeUnit.MILLISECONDS; this.max = 0; + this.unitMax = TimeUnit.MILLISECONDS; this.mandatoryStop = 0; + this.unitMandatoryStop = TimeUnit.MILLISECONDS; this.clock = Clock.systemDefaultZone(); } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BackoffTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java similarity index 99% rename from pulsar-client/src/test/java/org/apache/pulsar/client/impl/BackoffTest.java rename to pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java index 7f13acb7694..b3786236a70 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BackoffTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.pulsar.client.impl; +package org.apache.pulsar.common.util; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java index 4da6b7998a0..b6b5c57ccea 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LockManagerImpl.java @@ -27,7 +27,7 @@ import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.common.util.FutureUtil; @@ -52,7 +52,7 @@ class LockManagerImpl<T> implements LockManager<T> { private final MetadataCache<T> cache; private final MetadataSerde<T> serde; private final FutureUtil.Sequencer<Void> sequencer; - private final ExecutorService executor; + private final ScheduledExecutorService executor; private enum State { Ready, Closed @@ -60,13 +60,13 @@ class LockManagerImpl<T> implements LockManager<T> { private State state = State.Ready; - LockManagerImpl(MetadataStoreExtended store, Class<T> clazz, ExecutorService executor) { + LockManagerImpl(MetadataStoreExtended store, Class<T> clazz, ScheduledExecutorService executor) { this(store, new JSONMetadataSerdeSimpleType<>( TypeFactory.defaultInstance().constructSimpleType(clazz, null)), executor); } - LockManagerImpl(MetadataStoreExtended store, MetadataSerde<T> serde, ExecutorService executor) { + LockManagerImpl(MetadataStoreExtended store, MetadataSerde<T> serde, ScheduledExecutorService executor) { this.store = store; this.cache = store.getMetadataCache(serde); this.serde = serde; @@ -83,7 +83,7 @@ class LockManagerImpl<T> implements LockManager<T> { @Override public CompletableFuture<ResourceLock<T>> acquireLock(String path, T value) { - ResourceLockImpl<T> lock = new ResourceLockImpl<>(store, serde, path); + ResourceLockImpl<T> lock = new ResourceLockImpl<>(store, serde, path, executor); CompletableFuture<ResourceLock<T>> result = new CompletableFuture<>(); lock.acquire(value).thenRun(() -> { diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java index 93c994b2436..692f224594c 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java @@ -21,8 +21,13 @@ package org.apache.pulsar.metadata.coordination.impl; import java.util.EnumSet; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.common.concurrent.FutureUtils; +import org.apache.pulsar.common.util.Backoff; +import org.apache.pulsar.common.util.BackoffBuilder; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.MetadataSerde; @@ -44,7 +49,10 @@ public class ResourceLockImpl<T> implements ResourceLock<T> { private long version; private final CompletableFuture<Void> expiredFuture; private boolean revalidateAfterReconnection = false; + private final Backoff backoff; private final FutureUtil.Sequencer<Void> sequencer; + private final ScheduledExecutorService executor; + private ScheduledFuture<?> revalidateTask; private enum State { Init, @@ -55,7 +63,8 @@ public class ResourceLockImpl<T> implements ResourceLock<T> { private State state; - public ResourceLockImpl(MetadataStoreExtended store, MetadataSerde<T> serde, String path) { + ResourceLockImpl(MetadataStoreExtended store, MetadataSerde<T> serde, String path, + ScheduledExecutorService executor) { this.store = store; this.serde = serde; this.path = path; @@ -63,6 +72,11 @@ public class ResourceLockImpl<T> implements ResourceLock<T> { this.expiredFuture = new CompletableFuture<>(); this.sequencer = FutureUtil.Sequencer.create(); this.state = State.Init; + this.executor = executor; + this.backoff = new BackoffBuilder() + .setInitialTime(100, TimeUnit.MILLISECONDS) + .setMax(60, TimeUnit.SECONDS) + .create(); } @Override @@ -93,6 +107,10 @@ public class ResourceLockImpl<T> implements ResourceLock<T> { } state = State.Releasing; + if (revalidateTask != null) { + revalidateTask.cancel(true); + } + CompletableFuture<Void> result = new CompletableFuture<>(); store.delete(path, Optional.of(version)) @@ -210,8 +228,15 @@ public class ResourceLockImpl<T> implements ResourceLock<T> { * This method is thread-safe and it will perform multiple re-validation operations in turn. */ synchronized CompletableFuture<Void> silentRevalidateOnce() { + if (state != State.Valid) { + return CompletableFuture.completedFuture(null); + } + return sequencer.sequential(() -> revalidate(value)) - .thenRun(() -> log.info("Successfully revalidated the lock on {}", path)) + .thenRun(() -> { + log.info("Successfully revalidated the lock on {}", path); + backoff.reset(); + }) .exceptionally(ex -> { synchronized (ResourceLockImpl.this) { Throwable realCause = FutureUtil.unwrapCompletionException(ex); @@ -225,8 +250,12 @@ public class ResourceLockImpl<T> implements ResourceLock<T> { // Continue assuming we hold the lock, until we can revalidate it, either // on Reconnected or SessionReestablished events. revalidateAfterReconnection = true; - log.warn("Failed to revalidate the lock at {}. Retrying later on reconnection {}", path, - realCause.getMessage()); + + long delayMillis = backoff.next(); + log.warn("Failed to revalidate the lock at {}: {} - Retrying in {} seconds", path, + realCause.getMessage(), delayMillis / 1000.0); + revalidateTask = + executor.schedule(this::silentRevalidateOnce, delayMillis, TimeUnit.MILLISECONDS); } } return null; diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java index 05e6d4a3845..ebd60bad550 100644 --- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java +++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java @@ -35,6 +35,7 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import lombok.Cleanup; import org.apache.pulsar.common.util.ObjectMapperFactory; +import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.MetadataCache; import org.apache.pulsar.metadata.api.MetadataStoreConfig; import org.apache.pulsar.metadata.api.MetadataStoreException.LockBusyException; @@ -352,4 +353,34 @@ public class LockManagerTest extends BaseMetadataStoreTest { } }); } + + @Test(dataProvider = "impl") + public void lockDeletedAndReacquired(String provider, Supplier<String> urlSupplier) throws Exception { + @Cleanup + MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(), + MetadataStoreConfig.builder().fsyncEnable(false).build()); + + MetadataCache<String> cache = store.getMetadataCache(String.class); + + @Cleanup + CoordinationService coordinationService = new CoordinationServiceImpl(store); + + @Cleanup + LockManager<String> lockManager = coordinationService.getLockManager(String.class); + + String key = newKey(); + ResourceLock<String> lock = lockManager.acquireLock(key, "lock").join(); + assertEquals(lock.getValue(), "lock"); + var res = cache.get(key).join(); + assertTrue(res.isPresent()); + assertEquals(res.get(), "lock"); + + store.delete(key, Optional.empty()).join(); + + Awaitility.await().untilAsserted(() -> { + Optional<GetResult> val = store.get(key).join(); + assertTrue(val.isPresent()); + assertFalse(lock.getLockExpiredFuture().isDone()); + }); + } }