Use AtomicLongMap to calculate attempts

Author: cizezsy

extended/src/main/java/io/kubernetes/client/extended/workqueue/ratelimiter/BucketRateLimiter.java

@@ -30,7 +30,7 @@ public BucketRateLimiter(long capacity, long tokensGeneratedInPeriod, Duration p
   }
 
   @Override
-  public synchronized Duration when(T item) {
+  public Duration when(T item) {
     DelayGetter delayGetter = new DelayGetter();
     bucket.asAsyncScheduler().consume(1, delayGetter).complete(null);
     return delayGetter.getDelay();

extended/src/main/java/io/kubernetes/client/extended/workqueue/ratelimiter/ItemExponentialFailureRateLimiter.java

@@ -1,8 +1,7 @@
 package io.kubernetes.client.extended.workqueue.ratelimiter;
 
+import com.google.common.util.concurrent.AtomicLongMap;
 import java.time.Duration;
-import java.util.HashMap;
-import java.util.Map;
 
 /**
  * ItemExponentialFailureRateLimiter does a simple baseDelay*10<sup>num-failures</sup> limit dealing
@@ -13,30 +12,29 @@ public class ItemExponentialFailureRateLimiter<T> implements RateLimiter<T> {
   private Duration baseDelay;
   private Duration maxDelay;
 
-  private Map<T, Integer> failures;
+  private AtomicLongMap<T> failures;
 
   public ItemExponentialFailureRateLimiter(Duration baseDelay, Duration maxDelay) {
     this.baseDelay = baseDelay;
     this.maxDelay = maxDelay;
 
-    failures = new HashMap<>();
+    failures = AtomicLongMap.create();
   }
 
   @Override
-  public synchronized Duration when(T item) {
-    int exp = failures.getOrDefault(item, 0);
-    failures.put(item, exp + 1);
+  public Duration when(T item) {
+    long exp = failures.getAndIncrement(item);
     long d = maxDelay.toMillis() >> exp;
     return d > baseDelay.toMillis() ? baseDelay.multipliedBy(1 << exp) : maxDelay;
   }
 
   @Override
-  public synchronized void forget(T item) {
+  public void forget(T item) {
     failures.remove(item);
   }
 
   @Override
-  public synchronized int numRequeues(T item) {
-    return failures.getOrDefault(item, 0);
+  public int numRequeues(T item) {
+    return (int) failures.get(item);
   }
 }

extended/src/main/java/io/kubernetes/client/extended/workqueue/ratelimiter/ItemFastSlowRateLimiter.java

@@ -1,48 +1,46 @@
 package io.kubernetes.client.extended.workqueue.ratelimiter;
 
+import com.google.common.util.concurrent.AtomicLongMap;
 import java.time.Duration;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * ItemFastSlowRateLimiter does a quick retry for a certain number of attempts, then a slow retry
  * after that
  */
 public class ItemFastSlowRateLimiter<T> implements RateLimiter<T> {
 
-  private Map<T, Integer> failures;
-
   private Duration fastDelay;
   private Duration slowDelay;
   private int maxFastAttempts;
 
+  private AtomicLongMap<T> failures;
+
   public ItemFastSlowRateLimiter(Duration fastDelay, Duration slowDelay, int maxFastAttempts) {
     this.fastDelay = fastDelay;
     this.slowDelay = slowDelay;
     this.maxFastAttempts = maxFastAttempts;
 
-    failures = new HashMap<>();
+    failures = AtomicLongMap.create();
   }
 
   @Override
-  public synchronized Duration when(T item) {
-    int attempts = failures.getOrDefault(item, 0);
-    failures.put(item, attempts + 1);
-
-    if (attempts + 1 <= maxFastAttempts) {
+  public Duration when(T item) {
+    new ConcurrentHashMap<>().compute()
+    long attempts = failures.incrementAndGet(item);
+    if (attempts <= maxFastAttempts) {
       return fastDelay;
     }
-
     return slowDelay;
   }
 
   @Override
-  public synchronized void forget(T item) {
+  public void forget(T item) {
     failures.remove(item);
   }
 
   @Override
-  public synchronized int numRequeues(T item) {
-    return failures.getOrDefault(item, 0);
+  public int numRequeues(T item) {
+    return (int) failures.get(item);
   }
 }