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

@@ -17,6 +17,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -32,6 +33,7 @@
  * @author Costin Leau
  * @author Christoph Strobl
  * @author Ninad Divadkar
+ * @author Christian Bühler
  */
 class DefaultHashOperations<K, HK, HV> extends AbstractOperations<K, Object> implements HashOperations<K, HK, HV> {
 
@@ -141,16 +143,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

@@ -79,12 +79,15 @@
  * @author Ninad Divadkar
  * @author Anqing Shao
  * @author Mark Paluch
+ * @author Christian Bühler
  * @param <K> the Redis key type against which the template works (usually a String)
  * @param <V> the Redis value type against which the template works
  * @see StringRedisTemplate
  */
 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

@@ -38,6 +38,7 @@
  *
  * @author Costin Leau
  * @author Christoph Strobl
+ * @author Christian Bühler
  */
 public class DefaultRedisMap<K, V> implements RedisMap<K, V> {
 
@@ -155,20 +156,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

@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.Map;
@@ -62,6 +63,7 @@
  * @author Jennifer Hickey
  * @author Christoph Strobl
  * @author Thomas Darimont
+ * @author Christian Bühler
  */
 @RunWith(Parameterized.class)
 public abstract class AbstractRedisMapTests<K, V> {
@@ -396,6 +398,23 @@ public void testEntrySet() {
 		assertThat(values, not(hasItem(v2)));
 	}
 
+	@Test // DATAREDIS-803
+	@IfProfileValue(name = "runLongTests", value = "true")
+	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() {