DATAREDIS-803 - Work around Redis parameter limitation

Redis has a [limitation of 1024*1024
parameters](https://github.com/antirez/redis/blob/4.0.9/src/networking.c#L1200) for bulk operations.

To insert more than 1024*1024/2-1 entries with putAll(), they need to be
split up in multiple HMSET commands.

To reveive more than 1024*1024-1 entries with entrySet(), we can
directly use the HGETALL command instead of first fetching the keys with
HKEYS and then fetching the values with HMGET.

Author: christian-sahlmann

src/main/java/org/springframework/data/redis/core/DefaultHashOperations.java

@@ -15,13 +15,8 @@
  */
 package org.springframework.data.redis.core;
 
-import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.lang.Nullable;
@@ -141,16 +136,22 @@ public void putAll(K key, Map<? extends HK, ? extends HV> m) {
 
 		byte[] rawKey = rawKey(key);
 
-		Map<byte[], byte[]> hashes = new LinkedHashMap<>(m.size());
+		int size = Math.min(RedisTemplate.REDIS_MAX_ARGS / 2 - 1, m.size());
+		Map<byte[], byte[]> hashes = new LinkedHashMap<>(size);
 
-		for (Map.Entry<? extends HK, ? extends HV> entry : m.entrySet()) {
+		Iterator<? extends Entry<? extends HK, ? extends HV>> entries = m.entrySet().iterator();
+		while (entries.hasNext()) {
+			Entry<? extends HK, ? extends HV> entry = entries.next();
 			hashes.put(rawHashKey(entry.getKey()), rawHashValue(entry.getValue()));
-		}
 
-		execute(connection -> {
-			connection.hMSet(rawKey, hashes);
-			return null;
-		}, true);
+			if (!entries.hasNext() || hashes.size() == size) {
+				execute(connection -> {
+					connection.hMSet(rawKey, hashes);
+					return null;
+				}, true);
+				hashes.clear();
+			}
+		}
 	}
 
 	/*

src/main/java/org/springframework/data/redis/core/RedisTemplate.java

@@ -85,6 +85,8 @@
  */
 public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperations<K, V>, BeanClassLoaderAware {
 
+	public static final int REDIS_MAX_ARGS = 1024 * 1024;
+
 	private boolean enableTransactionSupport = false;
 	private boolean exposeConnection = false;
 	private boolean initialized = false;

src/main/java/org/springframework/data/redis/support/collections/DefaultRedisMap.java

@@ -155,20 +155,7 @@ public boolean containsValue(Object value) {
 	 */
 	@Override
 	public Set<java.util.Map.Entry<K, V>> entrySet() {
-
-		Set<K> keySet = keySet();
-		checkResult(keySet);
-		Collection<V> multiGet = hashOps.multiGet(keySet);
-
-		Iterator<K> keys = keySet.iterator();
-		Iterator<V> values = multiGet.iterator();
-
-		Set<Map.Entry<K, V>> entries = new LinkedHashSet<>();
-		while (keys.hasNext()) {
-			entries.add(new DefaultRedisMapEntry(keys.next(), values.next()));
-		}
-
-		return entries;
+		return hashOps.entries().entrySet();
 	}
 
 	/*

src/test/java/org/springframework/data/redis/support/collections/AbstractRedisMapTests.java

@@ -22,14 +22,8 @@
 
 import java.io.IOException;
 import java.text.DecimalFormat;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
 
 import org.junit.After;
 import org.junit.AfterClass;
@@ -396,6 +390,22 @@ public void testEntrySet() {
 		assertThat(values, not(hasItem(v2)));
 	}
 
+	@Test
+	public void testBigEntrySet() {
+		Set<Entry<K, V>> entries = map.entrySet();
+		assertTrue(entries.isEmpty());
+
+        Map<K, V> m = new HashMap<>();
+        for (int i = 0; i < RedisTemplate.REDIS_MAX_ARGS - 1; i++) {
+            m.put(getKey(), getValue());
+        }
+        map.putAll(m);
+
+		entries = map.entrySet();
+
+		assertEquals(RedisTemplate.REDIS_MAX_ARGS - 1, entries.size());
+	}
+
 	@Test
 	public void testPutIfAbsent() {