Add max retries and retry delay options to primitive/protocol configu…

…rations.
atomix · Dec 6, 2017 · 2d4ad14 · 2d4ad14
1 parent 100efcd
commit 2d4ad14
Show file tree

Hide file tree

Showing 44 changed files with 503 additions and 268 deletions.
diff --git a/core/src/main/java/io/atomix/counter/AtomicCounterBuilder.java b/core/src/main/java/io/atomix/counter/AtomicCounterBuilder.java
@@ -86,6 +86,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(Duration.ofSeconds(5))
         .withMaxTimeout(Duration.ofSeconds(30))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 
@@ -94,6 +96,8 @@ private PrimitiveProtocol newMultiPrimaryProtocol(Consistency consistency, Repli
         .withConsistency(consistency)
         .withReplication(replication)
         .withBackups(backups())
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/counter/impl/AtomicCounterProxyBuilder.java b/core/src/main/java/io/atomix/counter/impl/AtomicCounterProxyBuilder.java
@@ -43,8 +43,7 @@ public CompletableFuture<AtomicCounter> buildAsync() {
         .getPartitionGroup(protocol)
         .getPartition(name())
         .getPrimitiveClient()
-        .proxyBuilder(name(), primitiveType(), protocol)
-        .build()
+        .newProxy(name(), primitiveType(), protocol)
         .open()
         .thenApply(proxy -> new AtomicCounterProxy(proxy).sync());
   }

diff --git a/core/src/main/java/io/atomix/election/LeaderElectionBuilder.java b/core/src/main/java/io/atomix/election/LeaderElectionBuilder.java
@@ -134,6 +134,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(electionTimeout)
         .withMaxTimeout(Duration.ofSeconds(5))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/election/LeaderElectorBuilder.java b/core/src/main/java/io/atomix/election/LeaderElectorBuilder.java
@@ -134,6 +134,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(electionTimeout)
         .withMaxTimeout(Duration.ofSeconds(5))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/election/impl/LeaderElectionProxyBuilder.java b/core/src/main/java/io/atomix/election/impl/LeaderElectionProxyBuilder.java
@@ -43,8 +43,7 @@ public CompletableFuture<LeaderElection<T>> buildAsync() {
         .getPartitionGroup(protocol)
         .getPartition(name())
         .getPrimitiveClient()
-        .proxyBuilder(name(), primitiveType(), protocol)
-        .build()
+        .newProxy(name(), primitiveType(), protocol)
         .open()
         .thenApply(proxy -> new TranscodingAsyncLeaderElection<T, byte[]>(new LeaderElectionProxy(proxy), serializer()::encode, serializer()::decode).sync());
   }

diff --git a/core/src/main/java/io/atomix/election/impl/LeaderElectorPrimaryElectionService.java b/core/src/main/java/io/atomix/election/impl/LeaderElectorPrimaryElectionService.java
@@ -89,15 +89,14 @@ public void removeListener(PrimaryElectionEventListener listener) {
   @SuppressWarnings("unchecked")
   private AsyncLeaderElector<NodeId> newLeaderElector(Partition partition) {
     PrimitiveProxy proxy = partition.getPrimitiveClient()
-        .proxyBuilder(PRIMITIVE_NAME, LeaderElectorType.instance(), RaftProtocol.builder()
+        .newProxy(PRIMITIVE_NAME, LeaderElectorType.instance(), RaftProtocol.builder()
             .withMinTimeout(Duration.ofMillis(250))
             .withMaxTimeout(Duration.ofSeconds(5))
             .withReadConsistency(ReadConsistency.LINEARIZABLE)
             .withCommunicationStrategy(CommunicationStrategy.LEADER)
             .withRecoveryStrategy(RecoveryStrategy.RECOVER)
-            .build())
-        .withMaxRetries(5)
-        .build();
+            .withMaxRetries(5)
+            .build());
     AsyncLeaderElector<byte[]> leaderElector = new LeaderElectorProxy(proxy.open().join());
     return new TranscodingAsyncLeaderElector<>(leaderElector, SERIALIZER::encode, SERIALIZER::decode);
   }

diff --git a/core/src/main/java/io/atomix/election/impl/LeaderElectorProxyBuilder.java b/core/src/main/java/io/atomix/election/impl/LeaderElectorProxyBuilder.java
@@ -60,7 +60,7 @@ public CompletableFuture<LeaderElector<T>> buildAsync() {
     Map<PartitionId, CompletableFuture<AsyncLeaderElector<T>>> electors = Maps.newConcurrentMap();
     for (Partition partition : partitions.getPartitions()) {
       electors.put(partition.id(),
-          newLeaderElector(partition.getPrimitiveClient().proxyBuilder(name(), primitiveType(), protocol).build()));
+          newLeaderElector(partition.getPrimitiveClient().newProxy(name(), primitiveType(), protocol)));
     }
 
     Partitioner<String> partitioner = topic -> partitions.getPartition(topic).id();

diff --git a/core/src/main/java/io/atomix/generator/AtomicIdGeneratorBuilder.java b/core/src/main/java/io/atomix/generator/AtomicIdGeneratorBuilder.java
@@ -86,6 +86,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(Duration.ofSeconds(5))
         .withMaxTimeout(Duration.ofSeconds(30))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 
@@ -94,6 +96,8 @@ private PrimitiveProtocol newMultiPrimaryProtocol(Consistency consistency, Repli
         .withConsistency(consistency)
         .withReplication(replication)
         .withBackups(backups())
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/generator/impl/DelegatingIdGeneratorBuilder.java b/core/src/main/java/io/atomix/generator/impl/DelegatingIdGeneratorBuilder.java
@@ -44,8 +44,7 @@ public CompletableFuture<AtomicIdGenerator> buildAsync() {
         .getPartitionGroup(protocol)
         .getPartition(name())
         .getPrimitiveClient()
-        .proxyBuilder(name(), primitiveType(), protocol)
-        .build()
+        .newProxy(name(), primitiveType(), protocol)
         .open()
         .thenApply(proxy -> new DelegatingIdGenerator(new AtomicCounterProxy(proxy)).sync());
   }

diff --git a/core/src/main/java/io/atomix/generator/impl/IdGeneratorSessionIdService.java b/core/src/main/java/io/atomix/generator/impl/IdGeneratorSessionIdService.java
@@ -58,15 +58,14 @@ public CompletableFuture<SessionId> nextSessionId() {
   public CompletableFuture<SessionIdService> open() {
     PrimitiveProxy proxy = partitions.getPartition(PRIMITIVE_NAME)
         .getPrimitiveClient()
-        .proxyBuilder(PRIMITIVE_NAME, AtomicIdGeneratorType.instance(), RaftProtocol.builder()
+        .newProxy(PRIMITIVE_NAME, AtomicIdGeneratorType.instance(), RaftProtocol.builder()
             .withMinTimeout(Duration.ofMillis(250))
             .withMaxTimeout(Duration.ofSeconds(5))
             .withReadConsistency(ReadConsistency.LINEARIZABLE)
             .withCommunicationStrategy(CommunicationStrategy.LEADER)
             .withRecoveryStrategy(RecoveryStrategy.RECOVER)
-            .build())
-        .withMaxRetries(5)
-        .build();
+            .withMaxRetries(5)
+            .build());
     return proxy.open()
         .thenApply(v -> {
           idGenerator = new DelegatingIdGenerator(new AtomicCounterProxy(proxy));

diff --git a/core/src/main/java/io/atomix/lock/DistributedLockBuilder.java b/core/src/main/java/io/atomix/lock/DistributedLockBuilder.java
@@ -112,6 +112,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(lockTimeout)
         .withMaxTimeout(Duration.ofSeconds(5))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/lock/impl/DistributedLockProxyBuilder.java b/core/src/main/java/io/atomix/lock/impl/DistributedLockProxyBuilder.java
@@ -43,8 +43,7 @@ public CompletableFuture<DistributedLock> buildAsync() {
         .getPartitionGroup(protocol)
         .getPartition(name())
         .getPrimitiveClient()
-        .proxyBuilder(name(), primitiveType(), protocol)
-        .build()
+        .newProxy(name(), primitiveType(), protocol)
         .open()
         .thenApply(proxy -> new DistributedLockProxy(proxy).sync());
   }

diff --git a/core/src/main/java/io/atomix/map/AtomicCounterMapBuilder.java b/core/src/main/java/io/atomix/map/AtomicCounterMapBuilder.java
@@ -86,6 +86,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(Duration.ofSeconds(5))
         .withMaxTimeout(Duration.ofSeconds(30))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 
@@ -94,6 +96,8 @@ private PrimitiveProtocol newMultiPrimaryProtocol(Consistency consistency, Repli
         .withConsistency(consistency)
         .withReplication(replication)
         .withBackups(backups())
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/map/ConsistentMapBuilder.java b/core/src/main/java/io/atomix/map/ConsistentMapBuilder.java
@@ -111,6 +111,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(Duration.ofSeconds(5))
         .withMaxTimeout(Duration.ofSeconds(30))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 
@@ -119,6 +121,8 @@ private PrimitiveProtocol newMultiPrimaryProtocol(Consistency consistency, Repli
         .withConsistency(consistency)
         .withReplication(replication)
         .withBackups(backups())
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/map/ConsistentTreeMapBuilder.java b/core/src/main/java/io/atomix/map/ConsistentTreeMapBuilder.java
@@ -88,6 +88,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(Duration.ofSeconds(5))
         .withMaxTimeout(Duration.ofSeconds(30))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 
@@ -96,6 +98,8 @@ private PrimitiveProtocol newMultiPrimaryProtocol(Consistency consistency, Repli
         .withConsistency(consistency)
         .withReplication(replication)
         .withBackups(backups())
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/map/impl/AtomicCounterMapProxyBuilder.java b/core/src/main/java/io/atomix/map/impl/AtomicCounterMapProxyBuilder.java
@@ -45,8 +45,7 @@ public CompletableFuture<AtomicCounterMap<K>> buildAsync() {
         .getPartitionGroup(protocol)
         .getPartition(name())
         .getPrimitiveClient()
-        .proxyBuilder(name(), primitiveType(), protocol)
-        .build()
+        .newProxy(name(), primitiveType(), protocol)
         .open()
         .thenApply(proxy -> {
           AtomicCounterMapProxy rawMap = new AtomicCounterMapProxy(proxy);

diff --git a/core/src/main/java/io/atomix/map/impl/ConsistentMapProxyBuilder.java b/core/src/main/java/io/atomix/map/impl/ConsistentMapProxyBuilder.java
@@ -61,8 +61,7 @@ public CompletableFuture<ConsistentMap<K, V>> buildAsync() {
     Map<PartitionId, CompletableFuture<AsyncConsistentMap<byte[], byte[]>>> maps = Maps.newConcurrentMap();
     for (Partition partition : partitions.getPartitions()) {
       maps.put(partition.id(), partition.getPrimitiveClient()
-          .proxyBuilder(name(), primitiveType(), protocol)
-          .build()
+          .newProxy(name(), primitiveType(), protocol)
           .open()
           .thenApply(proxy -> new TranscodingAsyncConsistentMap<>(
               new ConsistentMapProxy(proxy),

diff --git a/core/src/main/java/io/atomix/map/impl/ConsistentTreeMapProxyBuilder.java b/core/src/main/java/io/atomix/map/impl/ConsistentTreeMapProxyBuilder.java
@@ -47,8 +47,7 @@ public CompletableFuture<ConsistentTreeMap<V>> buildAsync() {
         .getPartitionGroup(protocol)
         .getPartition(name())
         .getPrimitiveClient()
-        .proxyBuilder(name(), primitiveType(), protocol)
-        .build()
+        .newProxy(name(), primitiveType(), protocol)
         .open()
         .thenApply(proxy -> {
           ConsistentTreeMapProxy rawMap = new ConsistentTreeMapProxy(proxy);

diff --git a/core/src/main/java/io/atomix/multimap/ConsistentMultimapBuilder.java b/core/src/main/java/io/atomix/multimap/ConsistentMultimapBuilder.java
@@ -88,6 +88,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(Duration.ofSeconds(5))
         .withMaxTimeout(Duration.ofSeconds(30))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 
@@ -96,6 +98,8 @@ private PrimitiveProtocol newMultiPrimaryProtocol(Consistency consistency, Repli
         .withConsistency(consistency)
         .withReplication(replication)
         .withBackups(backups())
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/multimap/impl/ConsistentMultimapProxyBuilder.java b/core/src/main/java/io/atomix/multimap/impl/ConsistentMultimapProxyBuilder.java
@@ -47,8 +47,7 @@ public CompletableFuture<ConsistentMultimap<K, V>> buildAsync() {
         .getPartitionGroup(protocol)
         .getPartition(name())
         .getPrimitiveClient()
-        .proxyBuilder(name(), primitiveType(), protocol)
-        .build()
+        .newProxy(name(), primitiveType(), protocol)
         .open()
         .thenApply(proxy -> {
           AsyncConsistentMultimap<String, byte[]> rawMap = new ConsistentSetMultimapProxy(proxy);

diff --git a/core/src/main/java/io/atomix/queue/WorkQueueBuilder.java b/core/src/main/java/io/atomix/queue/WorkQueueBuilder.java
@@ -86,6 +86,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(Duration.ofSeconds(5))
         .withMaxTimeout(Duration.ofSeconds(30))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 
@@ -94,6 +96,8 @@ private PrimitiveProtocol newMultiPrimaryProtocol(Consistency consistency, Repli
         .withConsistency(consistency)
         .withReplication(replication)
         .withBackups(backups())
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/queue/impl/WorkQueueProxyBuilder.java b/core/src/main/java/io/atomix/queue/impl/WorkQueueProxyBuilder.java
@@ -43,8 +43,7 @@ public CompletableFuture<WorkQueue<E>> buildAsync() {
         .getPartitionGroup(protocol)
         .getPartition(name())
         .getPrimitiveClient()
-        .proxyBuilder(name(), primitiveType(), protocol)
-        .build()
+        .newProxy(name(), primitiveType(), protocol)
         .open()
         .thenApply(proxy -> new TranscodingAsyncWorkQueue<E, byte[]>(new WorkQueueProxy(proxy), serializer()::encode, serializer()::decode).sync());
   }

diff --git a/core/src/main/java/io/atomix/set/impl/DelegatingDistributedSetBuilder.java b/core/src/main/java/io/atomix/set/impl/DelegatingDistributedSetBuilder.java
@@ -24,6 +24,7 @@
 import io.atomix.set.DistributedSetBuilder;
 import io.atomix.utils.serializer.Serializer;
 
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -87,6 +88,18 @@ public DistributedSetBuilder<E> withBackups(int numBackups) {
     return this;
   }
 
+  @Override
+  public DistributedSetBuilder<E> withMaxRetries(int maxRetries) {
+    mapBuilder.withMaxRetries(maxRetries);
+    return this;
+  }
+
+  @Override
+  public DistributedSetBuilder<E> withRetryDelay(Duration retryDelay) {
+    mapBuilder.withRetryDelay(retryDelay);
+    return this;
+  }
+
   @Override
   public boolean readOnly() {
     return mapBuilder.readOnly();
@@ -132,6 +145,16 @@ public int backups() {
     return mapBuilder.backups();
   }
 
+  @Override
+  public int maxRetries() {
+    return mapBuilder.maxRetries();
+  }
+
+  @Override
+  public Duration retryDelay() {
+    return mapBuilder.retryDelay();
+  }
+
   @Override
   public CompletableFuture<DistributedSet<E>> buildAsync() {
     return mapBuilder.buildAsync()

diff --git a/core/src/main/java/io/atomix/tree/DocumentTreeBuilder.java b/core/src/main/java/io/atomix/tree/DocumentTreeBuilder.java
@@ -127,6 +127,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(Duration.ofSeconds(5))
         .withMaxTimeout(Duration.ofSeconds(30))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 
@@ -135,6 +137,8 @@ private PrimitiveProtocol newMultiPrimaryProtocol(Consistency consistency, Repli
         .withConsistency(consistency)
         .withReplication(replication)
         .withBackups(backups())
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/tree/impl/DocumentTreeProxyBuilder.java b/core/src/main/java/io/atomix/tree/impl/DocumentTreeProxyBuilder.java
@@ -59,8 +59,7 @@ public CompletableFuture<DocumentTree<V>> buildAsync() {
     Map<PartitionId, CompletableFuture<AsyncDocumentTree<V>>> trees = Maps.newConcurrentMap();
     for (Partition partition : partitions.getPartitions()) {
       trees.put(partition.id(), partition.getPrimitiveClient()
-          .proxyBuilder(name(), primitiveType(), protocol)
-          .build()
+          .newProxy(name(), primitiveType(), protocol)
           .open()
           .thenApply(proxy -> {
             DocumentTreeProxy rawTree = new DocumentTreeProxy(proxy);

diff --git a/core/src/main/java/io/atomix/value/AtomicValueBuilder.java b/core/src/main/java/io/atomix/value/AtomicValueBuilder.java
@@ -89,6 +89,8 @@ private PrimitiveProtocol newRaftProtocol(Consistency readConsistency) {
         .withMinTimeout(Duration.ofSeconds(5))
         .withMaxTimeout(Duration.ofSeconds(30))
         .withReadConsistency(readConsistency == Consistency.LINEARIZABLE ? ReadConsistency.LINEARIZABLE : ReadConsistency.SEQUENTIAL)
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 
@@ -97,6 +99,8 @@ private PrimitiveProtocol newMultiPrimaryProtocol(Consistency consistency, Repli
         .withConsistency(consistency)
         .withReplication(replication)
         .withBackups(backups())
+        .withMaxRetries(maxRetries())
+        .withRetryDelay(retryDelay())
         .build();
   }
 }
diff --git a/core/src/main/java/io/atomix/value/impl/AtomicValueProxyBuilder.java b/core/src/main/java/io/atomix/value/impl/AtomicValueProxyBuilder.java
@@ -45,8 +45,7 @@ public CompletableFuture<AtomicValue<V>> buildAsync() {
         .getPartitionGroup(protocol)
         .getPartition(name())
         .getPrimitiveClient()
-        .proxyBuilder(name(), primitiveType(), protocol)
-        .build()
+        .newProxy(name(), primitiveType(), protocol)
         .open()
         .thenApply(proxy -> {
           AtomicValueProxy value = new AtomicValueProxy(proxy);