From c20d9ca7423046e9fce8f6b2b5d1527d3b0522cd Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Mon, 22 Jan 2018 15:06:23 +0100 Subject: [PATCH] DATAREDIS-756 - Order results of partitioned multi-key commands by positional keys. We now order and assemble results of multi-key commands (e.g. MGET with cross-slot keys) that are executed on different nodes by positional keys to retain duplicate keys in the requested order and retain the server response. Previously duplicate keys were treated as set of keys and the response didn't match the requested keys. Original Pull Request: #303 --- .../connection/ClusterCommandExecutor.java | 225 +++++++++++++----- .../connection/ClusterConnectionTests.java | 6 + .../jedis/JedisClusterConnectionTests.java | 21 ++ .../LettuceClusterConnectionTests.java | 20 ++ 4 files changed, 217 insertions(+), 55 deletions(-) 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 3420eb63b..8f54d99d6 100644 --- a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java +++ b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java @@ -15,10 +15,16 @@ */ package org.springframework.data.redis.connection; +import lombok.AccessLevel; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.RequiredArgsConstructor; + import java.util.*; import java.util.Map.Entry; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +import java.util.stream.Collectors; import org.springframework.beans.factory.DisposableBean; import org.springframework.core.task.AsyncTaskExecutor; @@ -202,7 +208,6 @@ public class ClusterCommandExecutor implements DisposableBean { Map>> futures = new LinkedHashMap<>(); for (RedisClusterNode node : resolvedRedisClusterNodes) { - futures.put(new NodeExecution(node), executor.submit(() -> executeCommandOnSingleNode(callback, node))); } @@ -225,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; } } @@ -271,29 +283,28 @@ public class ClusterCommandExecutor implements DisposableBean { public MultiNodeResult executeMultiKeyCommand(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 (Entry> entry : nodeKeyMap.entrySet()) { + for (Entry entry : nodeKeyMap.entrySet()) { if (entry.getKey().isMaster()) { - for (byte[] key : entry.getValue()) { + for (PositionalKey key : entry.getValue()) { futures.put(new NodeExecution(entry.getKey(), key), - executor.submit(() -> executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key))); + executor.submit(() -> executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key.getBytes()))); } } } @@ -326,6 +337,7 @@ public class ClusterCommandExecutor implements DisposableBean { return this.topologyProvider.getTopology(); } + @Nullable private DataAccessException convertToDataAccessException(Exception e) { return exceptionTranslationStrategy.translate(e); } @@ -384,17 +396,22 @@ 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 @Nullable PositionalKey positionalKey; - NodeExecution(RedisClusterNode node, Object... args) { + NodeExecution(RedisClusterNode node) { + this(node, null); + } + + NodeExecution(RedisClusterNode node, @Nullable PositionalKey positionalKey) { this.node = node; - this.args = args; + this.positionalKey = positionalKey; } /** @@ -404,45 +421,18 @@ public class ClusterCommandExecutor implements DisposableBean { 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; } - } /** @@ -515,17 +505,25 @@ public class ClusterCommandExecutor implements DisposableBean { * {@link MultiNodeResult} 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 MultiNodeResult { 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}. */ @@ -551,15 +549,23 @@ public class ClusterCommandExecutor implements DisposableBean { */ public List resultsAsListSortBy(byte[]... keys) { - ArrayList> clone = new ArrayList<>(nodeResults); - clone.sort(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); + + return result.values().stream().map(tNodeResult -> tNodeResult.value).collect(Collectors.toList()); } /** * @param returnValue can be {@literal null}. - * @return can be {@litearl null}. + * @return can be {@literal null}. */ @Nullable public T getFirstNonNullNotEmptyOrDefault(@Nullable T returnValue) { @@ -609,5 +615,114 @@ public class ClusterCommandExecutor implements DisposableBean { 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 + */ + @Getter + @EqualsAndHashCode + @RequiredArgsConstructor(access = AccessLevel.PRIVATE) + static class PositionalKey { + + private final ByteArrayWrapper key; + private final int 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(); + } + } + + /** + * Mutable data structure to represent multiple {@link PositionalKey}s. + * + * @author Mark Paluch + * @since 2.0.3 + */ + @RequiredArgsConstructor(access = AccessLevel.PRIVATE) + static class PositionalKeys implements Iterable { + + private final List 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 0fd81455f..ff9ef0638 100644 --- a/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java @@ -300,9 +300,15 @@ public interface ClusterConnectionTests { // DATAREDIS-315 void mGetShouldReturnCorrectlyWhenKeysDoNotMapToSameSlot(); + // DATAREDIS-756 + void mGetShouldReturnMultipleSameKeysWhenKeysDoNotMapToSameSlot(); + // DATAREDIS-315 void mGetShouldReturnCorrectlyWhenKeysMapToSameSlot(); + // DATAREDIS-756 + void mGetShouldReturnMultipleSameKeysWhenKeysMapToSameSlot(); + // DATAREDIS-315 void mSetNXShouldReturnFalseIfNotAllKeysSet(); 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 c01ab0b8d..5009e28a7 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 @@ -1069,6 +1069,17 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests { assertThat(clusterConnection.mGet(KEY_1_BYTES, KEY_2_BYTES), contains(VALUE_1_BYTES, VALUE_2_BYTES)); } + @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 mGetShouldReturnCorrectlyWhenKeysMapToSameSlot() { @@ -1079,6 +1090,16 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests { contains(VALUE_1_BYTES, 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 mSetNXShouldReturnFalseIfNotAllKeysSet() { 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 57157c212..e561cdb3d 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 @@ -1038,6 +1038,16 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { assertThat(clusterConnection.mGet(KEY_1_BYTES, KEY_2_BYTES), contains(VALUE_1_BYTES, VALUE_2_BYTES)); } + @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 mGetShouldReturnCorrectlyWhenKeysMapToSameSlot() { @@ -1048,6 +1058,16 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { contains(VALUE_1_BYTES, VALUE_2_BYTES)); } + @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 mSetNXShouldReturnFalseIfNotAllKeysSet() {