diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java index 7d6c1b137..b0b4213e1 100644 --- a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java +++ b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2017 the original author or authors. + * Copyright 2015-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -15,20 +15,8 @@ */ package org.springframework.data.redis.connection; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.Comparator; -import java.util.HashMap; -import java.util.HashSet; -import java.util.LinkedHashMap; -import java.util.LinkedHashSet; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.Map.Entry; -import java.util.Random; -import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -86,7 +74,7 @@ public class ClusterCommandExecutor implements DisposableBean { * @param topologyProvider must not be {@literal null}. * @param resourceProvider must not be {@literal null}. * @param exceptionTranslation must not be {@literal null}. - * @param executor can be {@literal null}. + * @param executor can be {@literal null}. Defaulted to {@link ThreadPoolTaskExecutor}. */ public ClusterCommandExecutor(ClusterTopologyProvider topologyProvider, ClusterNodeResourceProvider resourceProvider, ExceptionTranslationStrategy exceptionTranslation, AsyncTaskExecutor executor) { @@ -107,7 +95,7 @@ public class ClusterCommandExecutor implements DisposableBean { * Run {@link ClusterCommandCallback} on a random node. * * @param cmd must not be {@literal null}. - * @return + * @return never {@literal null}. */ public NodeResult executeCommandOnArbitraryNode(ClusterCommandCallback cmd) { @@ -165,8 +153,8 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Lookup node from the topology. * - * @param node - * @return + * @param node must not be {@literal null}. + * @return never {@literal null}. * @throws IllegalArgumentException in case the node could not be resolved to a topology-known node */ private RedisClusterNode lookupNode(RedisClusterNode node) { @@ -180,8 +168,8 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Run {@link ClusterCommandCallback} on all reachable master nodes. * - * @param cmd - * @return + * @param cmd must not be {@literal null}. + * @return never {@literal null}. * @throws ClusterCommandExecutionFailureException */ public MulitNodeResult executeCommandOnAllNodes(final ClusterCommandCallback cmd) { @@ -189,9 +177,9 @@ public class ClusterCommandExecutor implements DisposableBean { } /** - * @param callback - * @param nodes - * @return + * @param callback must not be {@literal null}. + * @param nodes must not be {@literal null}. + * @return never {@literal null}. * @throws ClusterCommandExecutionFailureException * @throws IllegalArgumentException in case the node could not be resolved to a topology-known node */ @@ -214,7 +202,6 @@ public class ClusterCommandExecutor implements DisposableBean { Map>> futures = new LinkedHashMap>>(); for (final RedisClusterNode node : resolvedRedisClusterNodes) { - futures.put(new NodeExecution(node), executor.submit(new Callable>() { @Override @@ -243,23 +230,30 @@ public class ClusterCommandExecutor implements DisposableBean { if (!entry.getValue().isDone() && !entry.getValue().isCancelled()) { done = false; } else { + + NodeExecution execution = entry.getKey(); try { String futureId = ObjectUtils.getIdentityHexString(entry.getValue()); if (!saveGuard.contains(futureId)) { - result.add(entry.getValue().get()); + + if (execution.isPositional()) { + result.add(execution.getPositionalKey(), entry.getValue().get()); + } else { + result.add(entry.getValue().get()); + } saveGuard.add(futureId); } } catch (ExecutionException e) { RuntimeException ex = convertToDataAccessException((Exception) e.getCause()); - exceptions.put(entry.getKey().getNode(), ex != null ? ex : e.getCause()); + exceptions.put(execution.getNode(), ex != null ? ex : e.getCause()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); RuntimeException ex = convertToDataAccessException((Exception) e.getCause()); - exceptions.put(entry.getKey().getNode(), ex != null ? ex : e.getCause()); + exceptions.put(execution.getNode(), ex != null ? ex : e.getCause()); break; } } @@ -282,39 +276,39 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Run {@link MultiKeyClusterCommandCallback} with on a curated set of nodes serving one or more keys. * - * @param cmd - * @return + * @param cmd must not be {@literal null}. + * @return never {@literal null}. * @throws ClusterCommandExecutionFailureException */ public MulitNodeResult executeMuliKeyCommand(final MultiKeyClusterCommandCallback cmd, Iterable keys) { - Map> nodeKeyMap = new HashMap>(); + Map nodeKeyMap = new HashMap(); + int index = 0; for (byte[] key : keys) { for (RedisClusterNode node : getClusterTopology().getKeyServingNodes(key)) { if (nodeKeyMap.containsKey(node)) { - nodeKeyMap.get(node).add(key); + nodeKeyMap.get(node).append(PositionalKey.of(key, index++)); } else { - Set keySet = new LinkedHashSet(); - keySet.add(key); - nodeKeyMap.put(node, keySet); + nodeKeyMap.put(node, PositionalKeys.of(PositionalKey.of(key, index++))); } } } Map>> futures = new LinkedHashMap>>(); - for (final Entry> entry : nodeKeyMap.entrySet()) { + for (final Entry entry : nodeKeyMap.entrySet()) { if (entry.getKey().isMaster()) { - for (final byte[] key : entry.getValue()) { + for (final PositionalKey key : entry.getValue()) { + futures.put(new NodeExecution(entry.getKey(), key), executor.submit(new Callable>() { @Override public NodeResult call() throws Exception { - return executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key); + return executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key.getBytes()); } })); } @@ -387,7 +381,7 @@ public class ClusterCommandExecutor implements DisposableBean { * @param * @since 1.7 */ - public static interface ClusterCommandCallback { + public interface ClusterCommandCallback { S doInCluster(T client); } @@ -398,7 +392,7 @@ public class ClusterCommandExecutor implements DisposableBean { * @param native driver connection * @param */ - public static interface MultiKeyClusterCommandCallback { + public interface MultiKeyClusterCommandCallback { S doInCluster(T client, byte[] key); } @@ -407,69 +401,47 @@ public class ClusterCommandExecutor implements DisposableBean { * keys, involved. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ private static class NodeExecution { - private RedisClusterNode node; - private Object[] args; + private final RedisClusterNode node; + private final PositionalKey positionalKey; - public NodeExecution(RedisClusterNode node, Object... args) { + NodeExecution(RedisClusterNode node) { + this(node, null); + } + + NodeExecution(RedisClusterNode node, PositionalKey positionalKey) { this.node = node; - this.args = args; + this.positionalKey = positionalKey; } /** * Get the {@link RedisClusterNode} the execution happens on. */ - public RedisClusterNode getNode() { + RedisClusterNode getNode() { return node; } - /* - * (non-Javadoc) - * @see java.lang.Object#hashCode() + /** + * Get the {@link PositionalKey} of this execution. + * + * @since 2.0.3 */ - @Override - public int hashCode() { - - int result = ObjectUtils.nullSafeHashCode(node); - return result + ObjectUtils.nullSafeHashCode(args); + PositionalKey getPositionalKey() { + return positionalKey; } - /* - * (non-Javadoc) - * @see java.lang.Object#equals(java.lang.Object) - */ - @Override - public boolean equals(Object obj) { - - if (this == obj) { - return true; - } - - if (obj == null) { - return false; - } - - if (!(obj instanceof NodeExecution)) { - return false; - } - - NodeExecution that = (NodeExecution) obj; - - if (!ObjectUtils.nullSafeEquals(this.node, that.node)) { - return false; - } - - return ObjectUtils.nullSafeEquals(this.args, that.args); + boolean isPositional() { + return positionalKey != null; } - } /** - * {@link NodeResult} encapsules the actual value returned by a {@link ClusterCommandCallback} on a given + * {@link NodeResult} encapsulates the actual value returned by a {@link ClusterCommandCallback} on a given * {@link RedisClusterNode}. * * @author Christoph Strobl @@ -485,8 +457,8 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Create new {@link NodeResult}. * - * @param node - * @param value + * @param node must not be {@literal null}. + * @param value can be {@literal null}. */ public NodeResult(RedisClusterNode node, T value) { this(node, value, new byte[] {}); @@ -495,9 +467,9 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Create new {@link NodeResult}. * - * @param node - * @param value - * @parm key + * @param node must not be {@literal null}. + * @param value can be {@literal null}. + * @param key must not be {@literal null}. */ public NodeResult(RedisClusterNode node, T value, byte[] key) { @@ -537,17 +509,25 @@ public class ClusterCommandExecutor implements DisposableBean { * {@link MulitNodeResult} holds all {@link NodeResult} of a command executed on multiple {@link RedisClusterNode}. * * @author Christoph Strobl + * @author Mark Paluch * @param * @since 1.7 */ public static class MulitNodeResult { List> nodeResults = new ArrayList>(); + Map> positionalResults = new LinkedHashMap>(); private void add(NodeResult result) { nodeResults.add(result); } + private void add(PositionalKey key, NodeResult result) { + + positionalResults.put(key, result); + add(result); + } + /** * @return never {@literal null}. */ @@ -573,15 +553,28 @@ public class ClusterCommandExecutor implements DisposableBean { */ public List resultsAsListSortBy(byte[]... keys) { - ArrayList> clone = new ArrayList>(nodeResults); - Collections.sort(clone, new ResultByReferenceKeyPositionComparator(keys)); + if (positionalResults.isEmpty()) { - return toList(clone); + List> clone = new ArrayList>(nodeResults); + clone.sort(new ResultByReferenceKeyPositionComparator(keys)); + + return toList(clone); + } + + Map> result = new TreeMap>( + new ResultByKeyPositionComparator(keys)); + result.putAll(positionalResults); + + List finalResult = new ArrayList(result.size()); + for (NodeResult value : result.values()) { + finalResult.add(value.value); + } + return finalResult; } /** - * @param returnValue - * @return + * @param returnValue can be {@literal null}. + * @return can be {@literal null}. */ public T getFirstNonNullNotEmptyOrDefault(T returnValue) { @@ -615,19 +608,156 @@ public class ClusterCommandExecutor implements DisposableBean { * {@link Comparator} for sorting {@link NodeResult} by reference keys. * * @author Christoph Strobl + * @author Mark Paluch */ private static class ResultByReferenceKeyPositionComparator implements Comparator> { List reference; - public ResultByReferenceKeyPositionComparator(byte[]... keys) { + ResultByReferenceKeyPositionComparator(byte[]... keys) { reference = new ArrayList(new ByteArraySet(Arrays.asList(keys))); } @Override public int compare(NodeResult o1, NodeResult o2) { - return Integer.valueOf(reference.indexOf(o1.key)).compareTo(reference.indexOf(o2.key)); + return Integer.compare(reference.indexOf(o1.key), reference.indexOf(o2.key)); + } + } + + /** + * {@link Comparator} for sorting {@link PositionalKey} by external {@link PositionalKeys}. + * + * @author Mark Paluch + * @since 2.0.3 + */ + private static class ResultByKeyPositionComparator implements Comparator { + + PositionalKeys reference; + + ResultByKeyPositionComparator(byte[]... keys) { + reference = PositionalKeys.of(keys); + } + + @Override + public int compare(PositionalKey o1, PositionalKey o2) { + return Integer.compare(reference.indexOf(o1), reference.indexOf(o2)); } } } + + /** + * Value object representing a Redis key at a particular command position. + * + * @author Mark Paluch + * @since 2.0.3 + */ + + static class PositionalKey { + + private final ByteArrayWrapper key; + private final int position; + + public PositionalKey(ByteArrayWrapper key, int position) { + this.key = key; + this.position = position; + } + + public static PositionalKey of(byte[] key, int index) { + return new PositionalKey(new ByteArrayWrapper(key), index); + } + + /** + * @return binary key. + */ + public byte[] getBytes() { + return key.getArray(); + } + + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; + + PositionalKey that = (PositionalKey) o; + + if (position != that.position) + return false; + return key.equals(that.key); + } + + @Override + public int hashCode() { + int result = key.hashCode(); + result = 31 * result + position; + return result; + } + } + + /** + * Mutable data structure to represent multiple {@link PositionalKey}s. + * + * @author Mark Paluch + * @since 2.0.3 + */ + static class PositionalKeys implements Iterable { + + private final List keys; + + public PositionalKeys(List keys) { + this.keys = keys; + } + + /** + * Create an empty {@link PositionalKeys}. + */ + public static PositionalKeys empty() { + return new PositionalKeys(new ArrayList()); + } + + /** + * Create an {@link PositionalKeys} from {@code keys}. + */ + public static PositionalKeys of(byte[]... keys) { + + List result = new ArrayList(keys.length); + + for (int i = 0; i < keys.length; i++) { + result.add(PositionalKey.of(keys[i], i)); + } + + return new PositionalKeys(result); + } + + /** + * Create an {@link PositionalKeys} from {@link PositionalKey}s. + */ + public static PositionalKeys of(PositionalKey... keys) { + + PositionalKeys result = PositionalKeys.empty(); + result.append(keys); + + return result; + } + + /** + * Append {@link PositionalKey}s to this object. + */ + public void append(PositionalKey... keys) { + this.keys.addAll(Arrays.asList(keys)); + } + + /** + * @return index of the {@link PositionalKey}. + */ + public int indexOf(PositionalKey key) { + return keys.indexOf(key); + } + + @Override + public Iterator iterator() { + return keys.iterator(); + } + } } diff --git a/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java index 1bf559015..aed9667f5 100644 --- a/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java @@ -243,9 +243,15 @@ public interface ClusterConnectionTests { // DATAREDIS-315 void lSetShouldSetElementAtPositionCorrectly(); + // DATAREDIS-756 + void mGetShouldReturnMultipleSameKeysWhenKeysDoNotMapToSameSlot(); + // DATAREDIS-315 void lRemShouldRemoveElementAtPositionCorrectly(); + // DATAREDIS-756 + void mGetShouldReturnMultipleSameKeysWhenKeysMapToSameSlot(); + // DATAREDIS-315 void lPopShouldReturnElementCorrectly(); diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java index 610345498..0812a75e0 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java @@ -967,6 +967,17 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests { assertThat(clusterConnection.sIsMember(KEY_1_BYTES, JedisConverters.toBytes("foo")), is(false)); } + @Test // DATAREDIS-756 + public void mGetShouldReturnMultipleSameKeysWhenKeysDoNotMapToSameSlot() { + + nativeConnection.set(KEY_1_BYTES, VALUE_1_BYTES); + nativeConnection.set(KEY_2_BYTES, VALUE_2_BYTES); + nativeConnection.set(KEY_3_BYTES, VALUE_3_BYTES); + + List result = clusterConnection.mGet(KEY_1_BYTES, KEY_2_BYTES, KEY_3_BYTES, KEY_1_BYTES); + assertThat(result, contains(VALUE_1_BYTES, VALUE_2_BYTES, VALUE_3_BYTES, VALUE_1_BYTES)); + } + @Test // DATAREDIS-315 public void sInterShouldWorkForKeysMappingToSameSlot() { @@ -976,6 +987,16 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests { assertThat(clusterConnection.sInter(SAME_SLOT_KEY_1_BYTES, SAME_SLOT_KEY_2_BYTES), hasItem(VALUE_2_BYTES)); } + @Test // DATAREDIS-756 + public void mGetShouldReturnMultipleSameKeysWhenKeysMapToSameSlot() { + + nativeConnection.set(SAME_SLOT_KEY_1_BYTES, VALUE_1_BYTES); + nativeConnection.set(SAME_SLOT_KEY_2_BYTES, VALUE_2_BYTES); + + List result = clusterConnection.mGet(SAME_SLOT_KEY_1_BYTES, SAME_SLOT_KEY_2_BYTES, SAME_SLOT_KEY_1_BYTES); + assertThat(result, contains(VALUE_1_BYTES, VALUE_2_BYTES, VALUE_1_BYTES)); + } + @Test // DATAREDIS-315 public void sInterShouldWorkForKeysNotMappingToSameSlot() { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java index c28dfde69..6b6421309 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java @@ -892,6 +892,16 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { assertThat(nativeConnection.exists(KEY_2), is(true)); } + @Test // DATAREDIS-756 + public void mGetShouldReturnMultipleSameKeysWhenKeysDoNotMapToSameSlot() { + + nativeConnection.set(KEY_1, VALUE_1); + nativeConnection.set(KEY_2, VALUE_2); + + assertThat(clusterConnection.mGet(KEY_1_BYTES, KEY_2_BYTES, KEY_1_BYTES), + contains(VALUE_1_BYTES, VALUE_2_BYTES, VALUE_1_BYTES)); + } + @Test // DATAREDIS-315 public void bRPopLPushShouldWorkOnSameSlotKeys() { @@ -901,6 +911,16 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { assertThat(nativeConnection.exists(SAME_SLOT_KEY_2), is(true)); } + @Test // DATAREDIS-756 + public void mGetShouldReturnMultipleSameKeysWhenKeysMapToSameSlot() { + + nativeConnection.set(SAME_SLOT_KEY_1, VALUE_1); + nativeConnection.set(SAME_SLOT_KEY_2, VALUE_2); + + assertThat(clusterConnection.mGet(SAME_SLOT_KEY_1_BYTES, SAME_SLOT_KEY_2_BYTES, SAME_SLOT_KEY_1_BYTES), + contains(VALUE_1_BYTES, VALUE_2_BYTES, VALUE_1_BYTES)); + } + @Test // DATAREDIS-315 public void rPopLPushShouldWorkWhenKeysOnSameSlot() {