fix: addOrSet(serverGroup) writes through one client per server group

evcache-core/src/main/java/com/netflix/evcache/EVCacheInternalImpl.java

@@ -16,6 +16,7 @@
 import java.net.InetSocketAddress;
 import java.util.*;
 import java.util.concurrent.Future;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.stream.Collectors;
 
 /**
@@ -105,13 +106,7 @@ public EVCacheLatch addOrSet(boolean replaceItem, String key, CachedData value,
     }
 
     private EVCacheLatch addOrSet(boolean replaceItem, String key, CachedData value, int timeToLive, EVCacheLatch.Policy policy, List<String> serverGroups, List<String> destinationIps) throws EVCacheException {
-        Map<ServerGroup, List<EVCacheClient>> clientsByServerGroup = _pool.getAllInstancesByZone();
-
-        List<EVCacheClient> evCacheClients = clientsByServerGroup.entrySet().stream()
-                .filter(entry -> serverGroups.contains(entry.getKey().getName()))
-                .map(Map.Entry::getValue)
-                .flatMap(List::stream)
-                .collect(Collectors.toList());
+        List<EVCacheClient> evCacheClients = selectOneClientPerServerGroup(_pool.getAllInstancesByZone(), serverGroups);
 
         if (null != destinationIps && !destinationIps.isEmpty()) {
             // identify that evcache client whose primary node is the destination ip for the key being processed
@@ -135,6 +130,17 @@ private EVCacheLatch addOrSet(boolean replaceItem, String key, CachedData value,
         }
     }
 
+    // Each server group holds poolSize clients connected to the same nodes. Writing through all of them would send the
+    // same write poolSize times, so pick one per server group, as getEVCacheClientForWrite() does for regular writes.
+    static List<EVCacheClient> selectOneClientPerServerGroup(Map<ServerGroup, List<EVCacheClient>> clientsByServerGroup, List<String> serverGroups) {
+        return clientsByServerGroup.entrySet().stream()
+                .filter(entry -> serverGroups.contains(entry.getKey().getName()))
+                .map(Map.Entry::getValue)
+                .filter(clients -> !clients.isEmpty())
+                .map(clients -> clients.get(ThreadLocalRandom.current().nextInt(clients.size())))
+                .collect(Collectors.toList());
+    }
+
     public KeyHashedState isKeyHashed(String appName, String serverGroup) {
         PropertyRepository propertyRepository = _poolManager.getEVCacheConfig().getPropertyRepository();
         boolean isKeyHashedAtAppOrAsg = propertyRepository.get(serverGroup + ".hash.key", Boolean.class).orElseGet(appName + ".hash.key").orElse(false).get();

evcache-core/src/test/java/com/netflix/evcache/EVCacheInternalImplTest.java

@@ -0,0 +1,71 @@
+package com.netflix.evcache;
+
+import com.netflix.evcache.pool.EVCacheClient;
+import com.netflix.evcache.pool.ServerGroup;
+import org.testng.annotations.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+public class EVCacheInternalImplTest {
+
+    private static List<EVCacheClient> clients(int poolSize) {
+        List<EVCacheClient> clients = new ArrayList<>();
+        for (int i = 0; i < poolSize; i++) {
+            clients.add(mock(EVCacheClient.class));
+        }
+        return clients;
+    }
+
+    @Test
+    public void picksOneClientForTargetServerGroupRegardlessOfPoolSize() {
+        List<EVCacheClient> target = clients(6);
+        Map<ServerGroup, List<EVCacheClient>> clientsByServerGroup = new LinkedHashMap<>();
+        clientsByServerGroup.put(new ServerGroup("us-east-1c", "app-useast1c-v000"), clients(6));
+        clientsByServerGroup.put(new ServerGroup("us-east-1d", "app-useast1d-v001"), target);
+
+        for (int i = 0; i < 100; i++) {
+            List<EVCacheClient> selected = EVCacheInternalImpl.selectOneClientPerServerGroup(
+                    clientsByServerGroup, Collections.singletonList("app-useast1d-v001"));
+            assertEquals(selected.size(), 1);
+            assertTrue(target.contains(selected.get(0)));
+        }
+    }
+
+    @Test
+    public void picksOneClientPerTargetServerGroup() {
+        List<EVCacheClient> groupC = clients(3);
+        List<EVCacheClient> groupD = clients(3);
+        Map<ServerGroup, List<EVCacheClient>> clientsByServerGroup = new LinkedHashMap<>();
+        clientsByServerGroup.put(new ServerGroup("us-east-1c", "app-useast1c-v000"), groupC);
+        clientsByServerGroup.put(new ServerGroup("us-east-1d", "app-useast1d-v001"), groupD);
+        clientsByServerGroup.put(new ServerGroup("us-east-1e", "app-useast1e-v002"), clients(3));
+
+        List<EVCacheClient> selected = EVCacheInternalImpl.selectOneClientPerServerGroup(
+                clientsByServerGroup, Arrays.asList("app-useast1c-v000", "app-useast1d-v001"));
+
+        assertEquals(selected.size(), 2);
+        assertTrue(groupC.contains(selected.get(0)));
+        assertTrue(groupD.contains(selected.get(1)));
+    }
+
+    @Test
+    public void returnsEmptyWhenNoServerGroupMatchesOrHasClients() {
+        Map<ServerGroup, List<EVCacheClient>> clientsByServerGroup = new LinkedHashMap<>();
+        clientsByServerGroup.put(new ServerGroup("us-east-1c", "app-useast1c-v000"), clients(3));
+        clientsByServerGroup.put(new ServerGroup("us-east-1d", "app-useast1d-v001"), Collections.emptyList());
+
+        assertTrue(EVCacheInternalImpl.selectOneClientPerServerGroup(
+                clientsByServerGroup, Collections.singletonList("app-useast1e-v002")).isEmpty());
+        assertTrue(EVCacheInternalImpl.selectOneClientPerServerGroup(
+                clientsByServerGroup, Collections.singletonList("app-useast1d-v001")).isEmpty());
+    }
+}

evcache-core/src/test/java/test-suite.xml

@@ -3,6 +3,7 @@
   <test name="UnitTests">
     <classes>
       <class name="com.netflix.evcache.pool.NodeLocatorLookupTest" />
+      <class name="com.netflix.evcache.EVCacheInternalImplTest" />
       <class name="com.netflix.evcache.pool.EVCacheValueSerdeTest" />
       <class name="com.netflix.evcache.config.EVCacheTranscoderPropertiesTest" />
     </classes>