CachingAsyncConsistentMap: evict failed get responses from cache

AtomixLeaderElector: evict failed getLeadership responses from cache immediately

Change-Id: I084abbe8f4669b98f85bc60dc6af9240630b7ce0
(cherry picked from commit 77012446)

@@ -95,7 +95,12 @@ public class CachingAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMa
 
 
     @Override
     @Override
     public CompletableFuture<Versioned<V>> get(K key) {
     public CompletableFuture<Versioned<V>> get(K key) {
-        return cache.getUnchecked(key);
+        return cache.getUnchecked(key)
+                .whenComplete((r, e) -> {
+                    if (e != null) {
+                        cache.invalidate(key);
+                    }
+                });
     }
     }
 
 
     @Override
     @Override
@@ -133,11 +138,11 @@ public class CachingAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMa
     @Override
     @Override
     public CompletableFuture<Boolean> remove(K key, V value) {
     public CompletableFuture<Boolean> remove(K key, V value) {
         return super.remove(key, value)
         return super.remove(key, value)
-                    .whenComplete((r, e) -> {
-                        if (r) {
-                            cache.invalidate(key);
-                        }
-                    });
+                .whenComplete((r, e) -> {
+                    if (r) {
+                        cache.invalidate(key);
+                    }
+                });
     }
     }
 
 
     @Override
     @Override
@@ -153,7 +158,7 @@ public class CachingAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMa
     @Override
     @Override
     public CompletableFuture<Versioned<V>> replace(K key, V value) {
     public CompletableFuture<Versioned<V>> replace(K key, V value) {
         return super.replace(key, value)
         return super.replace(key, value)
-                    .whenComplete((r, e) -> cache.invalidate(key));
+                .whenComplete((r, e) -> cache.invalidate(key));
     }
     }
 
 
     @Override
     @Override

@@ -135,7 +135,12 @@ public class AtomixLeaderElector extends AbstractResource<AtomixLeaderElector>
 
 
     @Override
     @Override
     public CompletableFuture<Leadership> getLeadership(String topic) {
     public CompletableFuture<Leadership> getLeadership(String topic) {
-        return cache.getUnchecked(topic);
+        return cache.getUnchecked(topic)
+                .whenComplete((r, e) -> {
+                    if (e != null) {
+                        cache.invalidate(topic);
+                    }
+                });
     }
     }
 
 
     @Override
     @Override