2121import java .util .EnumSet ;
2222import java .util .Optional ;
2323import java .util .concurrent .CompletableFuture ;
24+ import java .util .concurrent .ScheduledExecutorService ;
25+ import java .util .concurrent .ScheduledFuture ;
26+ import java .util .concurrent .TimeUnit ;
2427import lombok .extern .slf4j .Slf4j ;
2528import org .apache .bookkeeper .common .concurrent .FutureUtils ;
29+ import org .apache .pulsar .common .util .Backoff ;
30+ import org .apache .pulsar .common .util .BackoffBuilder ;
2631import org .apache .pulsar .common .util .FutureUtil ;
2732import org .apache .pulsar .metadata .api .GetResult ;
2833import org .apache .pulsar .metadata .api .MetadataSerde ;
@@ -44,7 +49,10 @@ public class ResourceLockImpl<T> implements ResourceLock<T> {
4449 private long version ;
4550 private final CompletableFuture <Void > expiredFuture ;
4651 private boolean revalidateAfterReconnection = false ;
52+ private final Backoff backoff ;
4753 private final FutureUtil .Sequencer <Void > sequencer ;
54+ private final ScheduledExecutorService executor ;
55+ private ScheduledFuture <?> revalidateTask ;
4856
4957 private enum State {
5058 Init ,
@@ -55,14 +63,20 @@ private enum State {
5563
5664 private State state ;
5765
58- public ResourceLockImpl (MetadataStoreExtended store , MetadataSerde <T > serde , String path ) {
66+ ResourceLockImpl (MetadataStoreExtended store , MetadataSerde <T > serde , String path ,
67+ ScheduledExecutorService executor ) {
5968 this .store = store ;
6069 this .serde = serde ;
6170 this .path = path ;
6271 this .version = -1 ;
6372 this .expiredFuture = new CompletableFuture <>();
6473 this .sequencer = FutureUtil .Sequencer .create ();
6574 this .state = State .Init ;
75+ this .executor = executor ;
76+ this .backoff = new BackoffBuilder ()
77+ .setInitialTime (100 , TimeUnit .MILLISECONDS )
78+ .setMax (60 , TimeUnit .SECONDS )
79+ .create ();
6680 }
6781
6882 @ Override
@@ -93,6 +107,10 @@ public synchronized CompletableFuture<Void> release() {
93107 }
94108
95109 state = State .Releasing ;
110+ if (revalidateTask != null ) {
111+ revalidateTask .cancel (true );
112+ }
113+
96114 CompletableFuture <Void > result = new CompletableFuture <>();
97115
98116 store .delete (path , Optional .of (version ))
@@ -210,8 +228,15 @@ synchronized CompletableFuture<Void> revalidateIfNeededAfterReconnection() {
210228 * This method is thread-safe and it will perform multiple re-validation operations in turn.
211229 */
212230 synchronized CompletableFuture <Void > silentRevalidateOnce () {
231+ if (state != State .Valid ) {
232+ return CompletableFuture .completedFuture (null );
233+ }
234+
213235 return sequencer .sequential (() -> revalidate (value ))
214- .thenRun (() -> log .info ("Successfully revalidated the lock on {}" , path ))
236+ .thenRun (() -> {
237+ log .info ("Successfully revalidated the lock on {}" , path );
238+ backoff .reset ();
239+ })
215240 .exceptionally (ex -> {
216241 synchronized (ResourceLockImpl .this ) {
217242 Throwable realCause = FutureUtil .unwrapCompletionException (ex );
@@ -225,8 +250,12 @@ synchronized CompletableFuture<Void> silentRevalidateOnce() {
225250 // Continue assuming we hold the lock, until we can revalidate it, either
226251 // on Reconnected or SessionReestablished events.
227252 revalidateAfterReconnection = true ;
228- log .warn ("Failed to revalidate the lock at {}. Retrying later on reconnection {}" , path ,
229- realCause .getMessage ());
253+
254+ long delayMillis = backoff .next ();
255+ log .warn ("Failed to revalidate the lock at {}: {} - Retrying in {} seconds" , path ,
256+ realCause .getMessage (), delayMillis / 1000.0 );
257+ revalidateTask =
258+ executor .schedule (this ::silentRevalidateOnce , delayMillis , TimeUnit .MILLISECONDS );
230259 }
231260 }
232261 return null ;
0 commit comments