ISPN-5684 Make getAll work with compatibility mode in DIST

* When compatibility mode is enabled, a clustered getAll requests
  results in a local getAll, whose results should not be unboxed.
* Also, when unboxing the result back for the client, unboxing must
  happen regardless unless the value is null.

client/hotrod-client/src/test/java/org/infinispan/client/hotrod/GetAllCompatDistTest.java

@@ -0,0 +1,45 @@
+package org.infinispan.client.hotrod;
+
+import org.infinispan.client.hotrod.test.MultiHotRodServersTest;
+import org.infinispan.configuration.cache.CacheMode;
+import org.infinispan.configuration.cache.ConfigurationBuilder;
+import org.testng.annotations.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.testng.AssertJUnit.assertEquals;
+
+@Test(testName = "client.hotrod.GetAllCompatDistTest", groups = "functional")
+public class GetAllCompatDistTest extends MultiHotRodServersTest {
+
+   @Override
+   protected void createCacheManagers() throws Throwable {
+      createHotRodServers(2, getCacheConfiguration());
+   }
+
+   private ConfigurationBuilder getCacheConfiguration() {
+      ConfigurationBuilder builder = getDefaultClusteredCacheConfig(CacheMode.DIST_SYNC, false);
+      builder.clustering().hash().numOwners(1).compatibility().enabled(true);
+      return builder;
+   }
+
+   public void testGetAllWithCompatibility() {
+      RemoteCache<String, String> cache = client(0).getCache();
+      HashMap<String, String> cachedValues = new HashMap<>();
+      for(int i=0; i<100; i++){
+         String key = String.format("key-%d", i);
+         String value = String.format("value-%d", i);
+         cache.put(key, value);
+         cachedValues.put(key, value);
+      }
+
+      Map<String, String> values = cache.getAll(cachedValues.keySet());
+      assertEquals(cachedValues.size(), values.size());
+      for(String key : values.keySet()){
+         assertEquals(cachedValues.get(key), values.get(key));
+      }
+   }
+
+}
+

core/src/main/java/org/infinispan/interceptors/compat/BaseTypeConverterInterceptor.java

@@ -145,6 +145,10 @@ public Object visitGetAllCommand(InvocationContext ctx, GetAllCommand command) t
          command.setKeys(boxedKeys);
       }
       Object ret = invokeNextInterceptor(ctx, command);
+
+      if (ret != null && !needsUnboxing(ctx))
+         return ret;
+
       if (ret != null) {
          if (command.isReturnEntries()) {
             Map<Object, CacheEntry> map = (Map<Object, CacheEntry>) ret;
@@ -169,11 +173,7 @@ public Object visitGetAllCommand(InvocationContext ctx, GetAllCommand command) t
             Map<Object, Object> unboxed = command.createMap();
             for (Map.Entry<Object, Object> entry : map.entrySet()) {
                Object value = entry == null ? null : entry.getValue();
-               if (command.getRemotelyFetched() == null || !command.getRemotelyFetched().containsKey(entry.getKey())) {
-                  unboxed.put(converter.unboxKey(entry.getKey()), entry == null ? null : converter.unboxValue(value));
-               } else {
-                  unboxed.put(converter.unboxKey(entry.getKey()), value);
-               }
+               unboxed.put(converter.unboxKey(entry.getKey()), entry == null ? null : converter.unboxValue(value));
             }
             return unboxed;
          }