Fix: asyncBulkGet decoding of hashed and unhashed keys together in one request

evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java

@@ -1,29 +1,35 @@
 package com.netflix.evcache.test;
 
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
 import static org.testng.Assert.assertTrue;
 
+import com.netflix.evcache.EVCache;
+import com.netflix.evcache.EVCacheException;
+import com.netflix.evcache.EVCacheGetOperationListener;
+import com.netflix.evcache.EVCacheLatch;
+import com.netflix.evcache.operation.EVCacheOperationFuture;
+import com.netflix.evcache.pool.EVCacheClient;
+import com.netflix.evcache.pool.ServerGroup;
+import com.netflix.evcache.test.transcoder.Movie;
+import com.netflix.evcache.test.transcoder.MovieTranscoder;
+import com.netflix.evcache.util.KeyHasher;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-
-import java.util.*;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-
-import com.netflix.evcache.*;
-import com.netflix.evcache.pool.EVCacheClient;
-import com.netflix.evcache.pool.ServerGroup;
-import com.netflix.evcache.util.KeyHasher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.annotations.Test;
-
-import com.netflix.evcache.operation.EVCacheOperationFuture;
 import rx.schedulers.Schedulers;
 
-
-import static org.testng.Assert.*;
-
 public class EVCacheTestDI extends DIBase implements EVCacheGetOperationListener<String> {
     private static final Logger log = LoggerFactory.getLogger(EVCacheTestDI.class);
     private int loops = 1;
@@ -280,6 +286,8 @@ public void functionalTestsWithAppLevelAndASGLevelHashingScenarios() throws Exce
         assertTrue(manager.getEVCacheConfig().getPropertyRepository().get(appName + ".auto.hash.keys", Boolean.class).orElse(false).get());
         assertFalse(manager.getEVCacheConfig().getPropertyRepository().get(appName + ".hash.key", Boolean.class).orElse(false).get());
         testWithLargeKey();
+        testWithMixedKeys();
+        testWithMixedKeysAndCustomTranscoder();
         // negative scenario
         propertiesToSet.remove(appName + ".auto.hash.keys");
         refreshEVCache();
@@ -314,8 +322,13 @@ public void functionalTestsWithAppLevelAndASGLevelHashingScenarios() throws Exce
         }
         doFunctionalTests(true);
         for (ServerGroup serverGroup : clientsByServerGroup.keySet()) {
-            propertiesToSet.remove(serverGroup.getName());
+            propertiesToSet.remove(serverGroup.getName() + ".hash.key");
+            propertiesToSet.remove(serverGroup.getName() + ".hash.algo");
         }
+
+        propertiesToSet.remove(appName + ".hash.key", "true");
+
+        refreshEVCache();
     }
 
     private void testWithLargeKey() throws Exception {
@@ -330,8 +343,135 @@ private void testWithLargeKey() throws Exception {
         EVCacheLatch latch = evCache.set(key, value, EVCacheLatch.Policy.ALL);
         latch.await(1000, TimeUnit.MILLISECONDS);
 
+        String val = evCache.get(key);
+        // get
+        assertEquals(val, value);
+
+        // delete
+        Future<Boolean>[] futures = evCache.delete(key);
+        for (Future<Boolean> future : futures) {
+            future.get();
+        }
+    }
+
+    private void testWithMixedKeys() throws Exception {
+
+        EVCache[] evcacheInstance = new EVCache[2];
+        evcacheInstance[0] = getNewBuilder().setAppName(appName).setCachePrefix("cid").enableRetry().build();
+        evcacheInstance[1] = this.evCache;
+
+        Map<String, String> kv = new HashMap<>(6);
+        String oneLargeKey = null;
+        String oneSmallKey = null;
+        for (int k = 0; k < 3; k ++) {
+            StringBuilder sb = new StringBuilder();
+            sb.append("testWithSmallAndLargeKeysMixed");
+            for (int i= 0; i < 100; i++) {
+                sb.append(System.nanoTime());
+            }
+            oneLargeKey = sb.toString();
+            kv.put(oneLargeKey, UUID.randomUUID().toString());
+        }
+        for (int k = 3; k < 6; k ++) {
+            oneSmallKey = "testWithSmallAndLargeKeysMixed" + System.nanoTime();
+            kv.put(oneSmallKey, UUID.randomUUID().toString());
+        }
+
+        for (Map.Entry<String, String> entry : kv.entrySet()) {
+            EVCacheLatch latch = evCache.set(entry.getKey(), entry.getValue(), EVCacheLatch.Policy.ALL);
+            latch.await(10000, TimeUnit.MILLISECONDS);
+        }
+
         // get
-        assertEquals(evCache.get(key), value);
+        String val = evCache.get(oneLargeKey);
+        assertEquals(val, kv.get(oneLargeKey));
+        val = evCache.get(oneSmallKey);
+        assertEquals(val, kv.get(oneSmallKey));
+
+        // async bulk get
+        for (int op : new int[]{0, 1}) {
+            Map<String, String> results;
+            if (op == 0) {
+                CompletableFuture<Map<String, String>> future = evCache.getAsyncBulk(kv.keySet().toArray(new String[0]));
+                results = future.get(10000, TimeUnit.MILLISECONDS);
+            } else {
+                results = evCache.getBulk(kv.keySet().toArray(new String[0]));
+            }
+            assertEquals(results.size(), kv.size());
+            for (Map.Entry<String, String> result : results.entrySet()) {
+                assertEquals(result.getValue(), kv.get(result.getKey()));
+            }
+        }
+
+        // delete
+        for (Map.Entry<String, String> entry : kv.entrySet()) {
+            Future<Boolean>[] deleteFutures = evCache.delete(entry.getKey());
+            for (Future<Boolean> deleteFuture : deleteFutures) {
+                deleteFuture.get();
+            }
+        }
+
+    }
+
+    private void testWithMixedKeysAndCustomTranscoder() throws Exception {
+
+        com.netflix.evcache.EVCache evCache = getNewBuilder().setAppName(appName).setCachePrefix("cid").enableRetry()
+                .setTranscoder(new MovieTranscoder())
+                .build();
+
+        Map<String, Movie> kv = new HashMap<>(6);
+        String oneLargeKey = null;
+        String oneSmallKey = null;
+        for (int k = 0; k < 3; k ++) {
+            StringBuilder sb = new StringBuilder();
+            sb.append("testWithSmallAndLargeKeysMixed");
+            for (int i= 0; i < 100; i++) {
+                sb.append(System.nanoTime());
+            }
+            oneLargeKey = sb.toString();
+            kv.put(oneLargeKey, new Movie(k, String.valueOf(k)));
+        }
+        for (int k = 3; k < 6; k ++) {
+            oneSmallKey = "testWithSmallAndLargeKeysMixed" + System.nanoTime();
+            kv.put(oneSmallKey, new Movie(k, String.valueOf(k)));
+        }
+
+        for (Map.Entry<String, Movie> entry : kv.entrySet()) {
+            EVCacheLatch latch = evCache.set(entry.getKey(), entry.getValue(), EVCacheLatch.Policy.ALL);
+            latch.await(10000, TimeUnit.MILLISECONDS);
+        }
+
+        // get
+        Movie val = evCache.get(oneLargeKey);
+        assertEquals(val, kv.get(oneLargeKey));
+        val = evCache.get(oneSmallKey);
+        assertEquals(val, kv.get(oneSmallKey));
+
+        // async bulk get
+        for (int op : new int[]{0, 1}) {
+            Map<String, Movie> results = new HashMap<>();
+            if (op == 0) {
+                CompletableFuture<Map<String, Movie>> future = evCache.getAsyncBulk(kv.keySet().toArray(new String[0]));
+                results = future.get(10000, TimeUnit.MILLISECONDS);
+            // } else {
+                // TODO: getBulk api is known to be broken for un-hashed keys not decoding correctly when request contains both hashed and unhashed keys
+                // results = evCache.getBulk(kv.keySet().toArray(new String[0]));
+            }
+
+            for (Map.Entry<String, Movie> result : results.entrySet()) {
+                assertEquals(results.size(), kv.size());
+                assertEquals(result.getValue(), kv.get(result.getKey()), "Did not get the written value back with op " + (op == 0 ? "getAsyncBulk" : "getBulk"));
+            }
+        }
+
+        // delete
+        for (Map.Entry<String, Movie> entry : kv.entrySet()) {
+            Future<Boolean>[] deleteFutures = evCache.delete(entry.getKey());
+            for (Future<Boolean> deleteFuture : deleteFutures) {
+                deleteFuture.get();
+            }
+        }
+
     }
 
     private void doFunctionalTests(boolean isHashingEnabled) throws Exception {
@@ -458,6 +598,7 @@ public void testAll() {
                     testGetAndTouchObservable();
                     waitForCallbacks();
                     testAppendOrAdd();
+                    // functionalTestsWithAppLevelAndASGLevelHashingScenarios(); // slow, requires EVCache refresh
                     testTouch();
                     testDelete();
                     testInsert();

evcache-client/test/com/netflix/evcache/test/transcoder/Movie.java

@@ -0,0 +1,44 @@
+package com.netflix.evcache.test.transcoder;
+
+import java.util.Objects;
+
+public class Movie {
+    long id;
+    String name;
+
+    public Movie() { // required for decode
+    }
+
+    public Movie(long id, String name) {
+        this.id = id;
+        this.name = name;
+    }
+
+    public long getId() {
+        return id;
+    }
+
+    public void setId(long id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof Movie)) return false;
+        Movie movie = (Movie) o;
+        return id == movie.id && Objects.equals(name, movie.name);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(id, name);
+    }
+}

evcache-client/test/com/netflix/evcache/test/transcoder/MovieTranscoder.java

@@ -0,0 +1,44 @@
+package com.netflix.evcache.test.transcoder;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
+import net.spy.memcached.CachedData;
+import net.spy.memcached.transcoders.Transcoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MovieTranscoder implements Transcoder<Movie> {
+    private static final Logger LOGGER = LoggerFactory.getLogger(MovieTranscoder.class);
+    ObjectMapper mapper = new ObjectMapper();
+
+    @Override
+    public boolean asyncDecode(CachedData d) {
+        return false;
+    }
+
+    @Override
+    public CachedData encode(Movie movie) {
+        byte[] bytes;
+        try {
+            bytes = mapper.writeValueAsBytes(movie);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+        return new CachedData(0, bytes, bytes.length);
+    }
+
+    @Override
+    public Movie decode(CachedData d) {
+        try {
+            return mapper.readValue(d.getData(), Movie.class);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public int getMaxSize() {
+        return CachedData.MAX_SIZE;
+    }
+}