From f79d2921421f8ca1e58c93902e8f8ac5556bed41 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 | 314 +++++++++++++----- .../connection/ClusterConnectionTests.java | 6 + .../jedis/JedisClusterConnectionTests.java | 21 ++ .../LettuceClusterConnectionTests.java | 20 ++ 4 files changed, 269 insertions(+), 92 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 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() {