diff --git a/evcache-core/src/main/java/com/netflix/evcache/EVCacheInternalImpl.java b/evcache-core/src/main/java/com/netflix/evcache/EVCacheInternalImpl.java index 33257bb0..560e7613 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/EVCacheInternalImpl.java +++ b/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 serverGroups, List destinationIps) throws EVCacheException { - Map> clientsByServerGroup = _pool.getAllInstancesByZone(); - - List evCacheClients = clientsByServerGroup.entrySet().stream() - .filter(entry -> serverGroups.contains(entry.getKey().getName())) - .map(Map.Entry::getValue) - .flatMap(List::stream) - .collect(Collectors.toList()); + List 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 selectOneClientPerServerGroup(Map> clientsByServerGroup, List 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(); diff --git a/evcache-core/src/test/java/com/netflix/evcache/EVCacheInternalImplTest.java b/evcache-core/src/test/java/com/netflix/evcache/EVCacheInternalImplTest.java new file mode 100644 index 00000000..3bbb6ee4 --- /dev/null +++ b/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 clients(int poolSize) { + List clients = new ArrayList<>(); + for (int i = 0; i < poolSize; i++) { + clients.add(mock(EVCacheClient.class)); + } + return clients; + } + + @Test + public void picksOneClientForTargetServerGroupRegardlessOfPoolSize() { + List target = clients(6); + Map> 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 selected = EVCacheInternalImpl.selectOneClientPerServerGroup( + clientsByServerGroup, Collections.singletonList("app-useast1d-v001")); + assertEquals(selected.size(), 1); + assertTrue(target.contains(selected.get(0))); + } + } + + @Test + public void picksOneClientPerTargetServerGroup() { + List groupC = clients(3); + List groupD = clients(3); + Map> 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 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> 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()); + } +} diff --git a/evcache-core/src/test/java/test-suite.xml b/evcache-core/src/test/java/test-suite.xml index 39ab6875..6924a679 100644 --- a/evcache-core/src/test/java/test-suite.xml +++ b/evcache-core/src/test/java/test-suite.xml @@ -3,6 +3,7 @@ +