You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ar...@apache.org on 2016/11/30 17:15:53 UTC
[1/4] storm git commit: Added STORM-1886 to CHANGELOG.md
Repository: storm
Updated Branches:
refs/heads/master 152c51ab8 -> 285668742
Added STORM-1886 to CHANGELOG.md
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/28566874
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/28566874
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/28566874
Branch: refs/heads/master
Commit: 285668742742da6316d38c1ef492c7c873b5a649
Parents: 14098db
Author: Arun Mahadevan <ar...@apache.org>
Authored: Wed Nov 30 21:55:27 2016 +0530
Committer: Arun Mahadevan <ar...@apache.org>
Committed: Wed Nov 30 21:58:18 2016 +0530
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/28566874/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a9d4b30..c6396a9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 2.0.0
+ * STORM-1886: Extend KeyValueState iface with delete
* STORM-2193: Fix FilterConfiguration parameter order
* STORM-2209: Update documents adding new integration for some external systems
* STORM-2212: Remove Redundant Declarations in Maven POM Files
[3/4] storm git commit: STORM-1886 Address last comments
Posted by ar...@apache.org.
STORM-1886 Address last comments
* make tombstone static
* hide the use of Optional from the users
* fix race condition
* make encode and decode methods static
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/14098db4
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/14098db4
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/14098db4
Branch: refs/heads/master
Commit: 14098db42365b17faafe457faec1d9a0e14c7a13
Parents: 0d01d97
Author: Balazs Kossovics <ba...@s4m.io>
Authored: Mon Nov 28 17:24:16 2016 +0100
Committer: Arun Mahadevan <ar...@apache.org>
Committed: Wed Nov 30 21:58:18 2016 +0530
----------------------------------------------------------------------
.../storm/redis/state/RedisKeyValueState.java | 43 ++++++++++++--------
.../redis/state/RedisKeyValueStateTest.java | 2 +-
2 files changed, 26 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/14098db4/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
----------------------------------------------------------------------
diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java b/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
index db1a061..c2ca4df 100644
--- a/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
+++ b/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
@@ -43,13 +43,15 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
private static final Logger LOG = LoggerFactory.getLogger(RedisKeyValueState.class);
private static final String COMMIT_TXID_KEY = "commit";
private static final String PREPARE_TXID_KEY = "prepare";
+ private static final Serializer<Optional<byte[]>> internalValueSerializer = new DefaultStateSerializer<>();
+ private static final String TOMBSTONE = encode(internalValueSerializer.serialize(Optional.absent()));
private final String namespace;
private final String prepareNamespace;
private final String txidNamespace;
- private final String tombstone;
private final Serializer<K> keySerializer;
- private final Serializer<Optional<V>> valueSerializer;
+ private final Serializer<V> valueSerializer;
+
private final JedisCommandsInstanceContainer jedisContainer;
private Map<String, String> pendingPrepare;
private Map<String, String> pendingCommit;
@@ -60,21 +62,20 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
}
public RedisKeyValueState(String namespace, JedisPoolConfig poolConfig) {
- this(namespace, poolConfig, new DefaultStateSerializer<K>(), new DefaultStateSerializer<Optional<V>>());
+ this(namespace, poolConfig, new DefaultStateSerializer<K>(), new DefaultStateSerializer<V>());
}
- public RedisKeyValueState(String namespace, JedisPoolConfig poolConfig, Serializer<K> keySerializer, Serializer<Optional<V>> valueSerializer) {
+ public RedisKeyValueState(String namespace, JedisPoolConfig poolConfig, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
this(namespace, JedisCommandsContainerBuilder.build(poolConfig), keySerializer, valueSerializer);
}
public RedisKeyValueState(String namespace, JedisCommandsInstanceContainer jedisContainer,
- Serializer<K> keySerializer, Serializer<Optional<V>> valueSerializer) {
+ Serializer<K> keySerializer, Serializer<V> valueSerializer) {
this.namespace = namespace;
this.prepareNamespace = namespace + "$prepare";
this.txidNamespace = namespace + "$txid";
this.keySerializer = keySerializer;
this.valueSerializer = valueSerializer;
- this.tombstone = encode(valueSerializer.serialize(Optional.<V>absent()));
this.jedisContainer = jedisContainer;
this.pendingPrepare = new ConcurrentHashMap<>();
initTxids();
@@ -117,7 +118,7 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
LOG.debug("put key '{}', value '{}'", key, value);
String redisKey = encode(keySerializer.serialize(key));
pendingPrepare.put(redisKey,
- encode(valueSerializer.serialize(Optional.of(value))));
+ encode(internalValueSerializer.serialize(Optional.of(valueSerializer.serialize(value)))));
}
@Override
@@ -140,7 +141,12 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
}
V value = null;
if (redisValue != null) {
- value = valueSerializer.deserialize(decode(redisValue)).orNull();
+ Optional<byte[]> internalValue = internalValueSerializer.deserialize(decode(redisValue));
+ if (internalValue.isPresent()) {
+ value = valueSerializer.deserialize(internalValue.get());
+ } else {
+ value = null;
+ }
}
LOG.debug("Value for key '{}' is '{}'", key, value);
return value;
@@ -157,7 +163,7 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
LOG.debug("delete key '{}'", key);
String redisKey = encode(keySerializer.serialize(key));
V curr = get(key);
- pendingPrepare.put(redisKey, tombstone);
+ pendingPrepare.put(redisKey, TOMBSTONE);
return curr;
}
@@ -167,24 +173,25 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
validatePrepareTxid(txid);
JedisCommands commands = null;
try {
+ Map<String, String> currentPending = pendingPrepare;
+ pendingPrepare = new ConcurrentHashMap<>();
commands = jedisContainer.getInstance();
if (commands.exists(prepareNamespace)) {
LOG.debug("Prepared txn already exists, will merge", txid);
for (Map.Entry<String, String> e: pendingCommit.entrySet()) {
- if (!pendingPrepare.containsKey(e.getKey())) {
- pendingPrepare.put(e.getKey(), e.getValue());
+ if (!currentPending.containsKey(e.getKey())) {
+ currentPending.put(e.getKey(), e.getValue());
}
}
}
- if (!pendingPrepare.isEmpty()) {
- commands.hmset(prepareNamespace, pendingPrepare);
+ if (!currentPending.isEmpty()) {
+ commands.hmset(prepareNamespace, currentPending);
} else {
LOG.debug("Nothing to save for prepareCommit, txid {}.", txid);
}
txIds.put(PREPARE_TXID_KEY, String.valueOf(txid));
commands.hmset(txidNamespace, txIds);
- pendingCommit = Collections.unmodifiableMap(pendingPrepare);
- pendingPrepare = new ConcurrentHashMap<>();
+ pendingCommit = Collections.unmodifiableMap(currentPending);
} finally {
jedisContainer.returnInstance(commands);
}
@@ -201,7 +208,7 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
List<String> keysToDelete = new ArrayList<>();
Map<String, String> keysToAdd = new HashMap<>();
for(Map.Entry<String, String> entry: pendingCommit.entrySet()) {
- if (tombstone.equals(entry.getValue())) {
+ if (TOMBSTONE.equals(entry.getValue())) {
keysToDelete.add(entry.getKey());
} else {
keysToAdd.put(entry.getKey(), entry.getValue());
@@ -319,11 +326,11 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
return lastId;
}
- private String encode(byte[] bytes) {
+ private static String encode(byte[] bytes) {
return Base64.encodeBase64String(bytes);
}
- private byte[] decode(String s) {
+ private static byte[] decode(String s) {
return Base64.decodeBase64(s);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/14098db4/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
----------------------------------------------------------------------
diff --git a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
index 1bb72a7..e40885d 100644
--- a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
+++ b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
@@ -105,7 +105,7 @@ public class RedisKeyValueStateTest {
});
keyValueState = new RedisKeyValueState<String, String>("test", mockContainer, new DefaultStateSerializer<String>(),
- new DefaultStateSerializer<Optional<String>>());
+ new DefaultStateSerializer<String>());
}
[4/4] storm git commit: STORM-1886 Extend KeyValueState iface with
delete
Posted by ar...@apache.org.
STORM-1886 Extend KeyValueState iface with delete
The patch also provides implementation for delete in
RedisKeyValueState and InMemoryKeyValueState.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/0d65e319
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/0d65e319
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/0d65e319
Branch: refs/heads/master
Commit: 0d65e3193ec112cfded5433ee1c459ab5d0203e9
Parents: 152c51a
Author: Balazs Kossovics <ba...@s4m.io>
Authored: Thu Jun 2 14:32:06 2016 +0200
Committer: Arun Mahadevan <ar...@apache.org>
Committed: Wed Nov 30 21:58:18 2016 +0530
----------------------------------------------------------------------
.../storm/redis/state/RedisKeyValueState.java | 66 +++++++++++++++-----
.../redis/state/RedisKeyValueStateTest.java | 62 ++++++++++++++++--
.../storm/state/InMemoryKeyValueState.java | 5 ++
.../org/apache/storm/state/KeyValueState.java | 7 +++
4 files changed, 121 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/0d65e319/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
----------------------------------------------------------------------
diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java b/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
index 8769cb0..c686941 100644
--- a/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
+++ b/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
@@ -31,6 +31,8 @@ import org.apache.commons.codec.binary.Base64;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
+import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
/**
@@ -48,7 +50,9 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
private final Serializer<V> valueSerializer;
private final JedisCommandsInstanceContainer jedisContainer;
private Map<String, String> pendingPrepare;
+ private Set<String> pendingDeletePrepare;
private Map<String, String> pendingCommit;
+ private Set<String> pendingDeleteCommit;
private Map<String, String> txIds;
public RedisKeyValueState(String namespace) {
@@ -71,7 +75,8 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
this.keySerializer = keySerializer;
this.valueSerializer = valueSerializer;
this.jedisContainer = jedisContainer;
- this.pendingPrepare = new ConcurrentHashMap<>();
+ this.pendingPrepare = new HashMap<>();
+ this.pendingDeletePrepare = new HashSet<>();
initTxids();
initPendingCommit();
}
@@ -98,9 +103,11 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
if (commands.exists(prepareNamespace)) {
LOG.debug("Loading previously prepared commit from {}", prepareNamespace);
pendingCommit = Collections.unmodifiableMap(commands.hgetAll(prepareNamespace));
+ pendingDeleteCommit = Collections.emptySet();
} else {
LOG.debug("No previously prepared commits.");
pendingCommit = Collections.emptyMap();
+ pendingDeleteCommit = Collections.emptySet();
}
} finally {
jedisContainer.returnInstance(commands);
@@ -110,8 +117,12 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
@Override
public void put(K key, V value) {
LOG.debug("put key '{}', value '{}'", key, value);
- pendingPrepare.put(encode(keySerializer.serialize(key)),
- encode(valueSerializer.serialize(value)));
+ String redisKey = encode(keySerializer.serialize(key));
+ synchronized (this) {
+ pendingPrepare.put(redisKey,
+ encode(valueSerializer.serialize(value)));
+ pendingDeletePrepare.remove(redisKey);
+ }
}
@Override
@@ -119,11 +130,17 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
LOG.debug("get key '{}'", key);
String redisKey = encode(keySerializer.serialize(key));
String redisValue = null;
- if (pendingPrepare.containsKey(redisKey)) {
- redisValue = pendingPrepare.get(redisKey);
- } else if (pendingCommit.containsKey(redisKey)) {
- redisValue = pendingCommit.get(redisKey);
- } else {
+ boolean found = false;
+ synchronized (this) {
+ if (pendingPrepare.containsKey(redisKey) || pendingDeletePrepare.contains(redisKey)) {
+ redisValue = pendingPrepare.get(redisKey);
+ found = true;
+ } else if (pendingCommit.containsKey(redisKey) || pendingDeleteCommit.contains(redisKey)) {
+ redisValue = pendingCommit.get(redisKey);
+ found = true;
+ }
+ }
+ if (!found) {
JedisCommands commands = null;
try {
commands = jedisContainer.getInstance();
@@ -147,7 +164,17 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
}
@Override
- public void prepareCommit(long txid) {
+ public void delete(K key) {
+ LOG.debug("delete key '{}'", key);
+ String redisKey = encode(keySerializer.serialize(key));
+ synchronized (this) {
+ pendingDeletePrepare.add(redisKey);
+ pendingPrepare.remove(redisKey);
+ }
+ }
+
+ @Override
+ public void prepareCommit(long txid){
LOG.debug("prepareCommit txid {}", txid);
validatePrepareTxid(txid);
JedisCommands commands = null;
@@ -161,15 +188,20 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
}
}
}
- if (!pendingPrepare.isEmpty()) {
+ if (!pendingPrepare.isEmpty() || !pendingDeletePrepare.isEmpty()) {
commands.hmset(prepareNamespace, pendingPrepare);
+ commands.hdel(prepareNamespace, pendingDeletePrepare.toArray(new String[pendingDeletePrepare.size()]));
} else {
LOG.debug("Nothing to save for prepareCommit, txid {}.", txid);
}
txIds.put(PREPARE_TXID_KEY, String.valueOf(txid));
commands.hmset(txidNamespace, txIds);
- pendingCommit = Collections.unmodifiableMap(pendingPrepare);
+ synchronized (this) {
+ pendingCommit = Collections.unmodifiableMap(pendingPrepare);
+ pendingDeleteCommit = Collections.unmodifiableSet(pendingDeletePrepare);
+ }
pendingPrepare = new ConcurrentHashMap<>();
+ pendingDeletePrepare = Collections.synchronizedSet(new HashSet<String>());
} finally {
jedisContainer.returnInstance(commands);
}
@@ -182,8 +214,9 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
JedisCommands commands = null;
try {
commands = jedisContainer.getInstance();
- if (!pendingCommit.isEmpty()) {
+ if (!pendingCommit.isEmpty() || !pendingDeleteCommit.isEmpty()) {
commands.hmset(namespace, pendingCommit);
+ commands.hdel(namespace, pendingDeleteCommit.toArray(new String[pendingDeleteCommit.size()]));
} else {
LOG.debug("Nothing to save for commit, txid {}.", txid);
}
@@ -191,6 +224,7 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
commands.hmset(txidNamespace, txIds);
commands.del(prepareNamespace);
pendingCommit = Collections.emptyMap();
+ pendingDeleteCommit = Collections.emptySet();
} finally {
jedisContainer.returnInstance(commands);
}
@@ -201,12 +235,14 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
JedisCommands commands = null;
try {
commands = jedisContainer.getInstance();
- if (!pendingPrepare.isEmpty()) {
+ if (!pendingPrepare.isEmpty() || !pendingDeletePrepare.isEmpty()) {
commands.hmset(namespace, pendingPrepare);
+ commands.hdel(namespace, pendingDeletePrepare.toArray(new String[pendingDeletePrepare.size()]));
} else {
LOG.debug("Nothing to save for commit");
}
- pendingPrepare = new ConcurrentHashMap<>();
+ pendingPrepare = new HashMap<>();
+ pendingDeletePrepare = new HashSet<>();
} finally {
jedisContainer.returnInstance(commands);
}
@@ -234,7 +270,9 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
commands.hmset(txidNamespace, txIds);
}
pendingCommit = Collections.emptyMap();
+ pendingDeleteCommit = Collections.emptySet();
pendingPrepare = new ConcurrentHashMap<>();
+ pendingDeletePrepare = Collections.synchronizedSet(new HashSet<String>());
} finally {
jedisContainer.returnInstance(commands);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/0d65e319/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
----------------------------------------------------------------------
diff --git a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
index ea8cc15..f5525d0 100644
--- a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
+++ b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
@@ -32,10 +32,9 @@ import redis.clients.jedis.SortingParams;
import redis.clients.jedis.Tuple;
import java.util.HashMap;
-import java.util.List;
+import java.util.Arrays;
import java.util.Map;
import java.util.Set;
-
import static org.junit.Assert.*;
/**
@@ -93,6 +92,17 @@ public class RedisKeyValueStateTest {
}
});
+ Mockito.when(mockCommands.hdel(Mockito.anyString(), Mockito.<String>anyVararg()))
+ .thenAnswer(new Answer<Long>() {
+ @Override
+ public Long answer(InvocationOnMock invocation) throws Throwable {
+ Object[] args = invocation.getArguments();
+ int argsSize = args.length;
+ String[] fields = Arrays.asList(args).subList(1, argsSize).toArray(new String[argsSize - 1]);
+ return hdel(mockMap, (String) args[0], fields);
+ }
+ });
+
keyValueState = new RedisKeyValueState<String, String>("test", mockContainer, new DefaultStateSerializer<String>(),
new DefaultStateSerializer<String>());
}
@@ -108,6 +118,19 @@ public class RedisKeyValueStateTest {
}
@Test
+ public void testPutAndDelete() throws Exception {
+ keyValueState.put("a", "1");
+ keyValueState.put("b", "2");
+ assertEquals("1", keyValueState.get("a"));
+ assertEquals("2", keyValueState.get("b"));
+ assertEquals(null, keyValueState.get("c"));
+ keyValueState.delete("a");
+ assertEquals(null, keyValueState.get("a"));
+ assertEquals("2", keyValueState.get("b"));
+ assertEquals(null, keyValueState.get("c"));
+ }
+
+ @Test
public void testPrepareCommitRollback() throws Exception {
keyValueState.put("a", "1");
keyValueState.put("b", "2");
@@ -124,6 +147,20 @@ public class RedisKeyValueStateTest {
assertArrayEquals(new String[]{"1", "2", "3"}, getValues());
keyValueState.rollback();
assertArrayEquals(new String[]{"1", "2", null}, getValues());
+ keyValueState.put("c", "3");
+ keyValueState.delete("b");
+ keyValueState.delete("c");
+ assertArrayEquals(new String[]{"1", null, null}, getValues());
+ keyValueState.prepareCommit(2);
+ assertArrayEquals(new String[]{"1", null, null}, getValues());
+ keyValueState.commit(2);
+ assertArrayEquals(new String[]{"1", null, null}, getValues());
+ keyValueState.put("b", "2");
+ keyValueState.prepareCommit(3);
+ keyValueState.put("c", "3");
+ assertArrayEquals(new String[]{"1", "2", "3"}, getValues());
+ keyValueState.rollback();
+ assertArrayEquals(new String[]{"1", null, null}, getValues());
}
private String[] getValues() {
@@ -135,13 +172,20 @@ public class RedisKeyValueStateTest {
}
private String hmset(Map<String, Map<String, String>> mockMap, String key, Map value) {
- mockMap.put(key, value);
+ Map<String, String> currentValue = mockMap.get(key);
+ if (currentValue == null) {
+ currentValue = new HashMap<>();
+ }
+ currentValue.putAll(value);
+ mockMap.put(key, currentValue);
return "";
}
private Long del(Map<String, Map<String, String>> mockMap, String key) {
- mockMap.remove(key);
- return 0L;
+ if (mockMap.remove(key) == null)
+ return 0L;
+ else
+ return 1L;
}
private String hget(Map<String, Map<String, String>> mockMap, String namespace, String key) {
@@ -151,4 +195,12 @@ public class RedisKeyValueStateTest {
return null;
}
+ private Long hdel(Map<String, Map<String, String>> mockMap, String namespace, String ... keys) {
+ Long count = 0L;
+ for (String key: keys) {
+ if (mockMap.get(namespace).remove(key) != null) count++;
+ }
+ return count;
+ }
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/0d65e319/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java b/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java
index 4b116ba..4774d72 100644
--- a/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java
+++ b/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java
@@ -68,6 +68,11 @@ public class InMemoryKeyValueState<K, V> implements KeyValueState<K, V> {
}
@Override
+ public void delete(K key) {
+ state.remove(key);
+ }
+
+ @Override
public void commit() {
commitedState = new TxIdState<>(DEFAULT_TXID, new ConcurrentHashMap<>(state));
}
http://git-wip-us.apache.org/repos/asf/storm/blob/0d65e319/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java b/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java
index 3ab60f1..0e1facb 100644
--- a/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java
+++ b/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java
@@ -45,4 +45,11 @@ public interface KeyValueState<K, V> extends State {
* @return the value or defaultValue if no mapping is found
*/
V get(K key, V defaultValue);
+
+ /**
+ * Deletes the value mapped to the key, if there is any
+ *
+ * @param key the key
+ */
+ void delete(K key);
}
[2/4] storm git commit: Use a Map with tombstones to represent
deletions
Posted by ar...@apache.org.
Use a Map with tombstones to represent deletions
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/0d01d974
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/0d01d974
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/0d01d974
Branch: refs/heads/master
Commit: 0d01d974bac1a3fd291c852339be7bc218cbc645
Parents: 0d65e31
Author: Balazs Kossovics <ba...@s4m.io>
Authored: Fri Jun 24 17:51:32 2016 +0200
Committer: Arun Mahadevan <ar...@apache.org>
Committed: Wed Nov 30 21:58:18 2016 +0530
----------------------------------------------------------------------
.../storm/redis/state/RedisKeyValueState.java | 95 +++++++++-----------
.../redis/state/RedisKeyValueStateTest.java | 9 +-
.../storm/state/DefaultStateSerializer.java | 2 +
.../storm/state/InMemoryKeyValueState.java | 4 +-
.../org/apache/storm/state/KeyValueState.java | 2 +-
.../storm/state/InMemoryKeyValueStateTest.java | 27 ++++++
6 files changed, 80 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/0d01d974/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
----------------------------------------------------------------------
diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java b/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
index c686941..db1a061 100644
--- a/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
+++ b/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java
@@ -17,6 +17,7 @@
*/
package org.apache.storm.redis.state;
+import com.google.common.base.Optional;
import org.apache.storm.state.DefaultStateSerializer;
import org.apache.storm.state.KeyValueState;
import org.apache.storm.state.Serializer;
@@ -31,8 +32,8 @@ import org.apache.commons.codec.binary.Base64;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
-import java.util.HashSet;
+import java.util.List;
+import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
/**
@@ -46,13 +47,12 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
private final String namespace;
private final String prepareNamespace;
private final String txidNamespace;
+ private final String tombstone;
private final Serializer<K> keySerializer;
- private final Serializer<V> valueSerializer;
+ private final Serializer<Optional<V>> valueSerializer;
private final JedisCommandsInstanceContainer jedisContainer;
private Map<String, String> pendingPrepare;
- private Set<String> pendingDeletePrepare;
private Map<String, String> pendingCommit;
- private Set<String> pendingDeleteCommit;
private Map<String, String> txIds;
public RedisKeyValueState(String namespace) {
@@ -60,23 +60,23 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
}
public RedisKeyValueState(String namespace, JedisPoolConfig poolConfig) {
- this(namespace, poolConfig, new DefaultStateSerializer<K>(), new DefaultStateSerializer<V>());
+ this(namespace, poolConfig, new DefaultStateSerializer<K>(), new DefaultStateSerializer<Optional<V>>());
}
- public RedisKeyValueState(String namespace, JedisPoolConfig poolConfig, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
+ public RedisKeyValueState(String namespace, JedisPoolConfig poolConfig, Serializer<K> keySerializer, Serializer<Optional<V>> valueSerializer) {
this(namespace, JedisCommandsContainerBuilder.build(poolConfig), keySerializer, valueSerializer);
}
public RedisKeyValueState(String namespace, JedisCommandsInstanceContainer jedisContainer,
- Serializer<K> keySerializer, Serializer<V> valueSerializer) {
+ Serializer<K> keySerializer, Serializer<Optional<V>> valueSerializer) {
this.namespace = namespace;
this.prepareNamespace = namespace + "$prepare";
this.txidNamespace = namespace + "$txid";
this.keySerializer = keySerializer;
this.valueSerializer = valueSerializer;
+ this.tombstone = encode(valueSerializer.serialize(Optional.<V>absent()));
this.jedisContainer = jedisContainer;
- this.pendingPrepare = new HashMap<>();
- this.pendingDeletePrepare = new HashSet<>();
+ this.pendingPrepare = new ConcurrentHashMap<>();
initTxids();
initPendingCommit();
}
@@ -103,11 +103,9 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
if (commands.exists(prepareNamespace)) {
LOG.debug("Loading previously prepared commit from {}", prepareNamespace);
pendingCommit = Collections.unmodifiableMap(commands.hgetAll(prepareNamespace));
- pendingDeleteCommit = Collections.emptySet();
} else {
LOG.debug("No previously prepared commits.");
pendingCommit = Collections.emptyMap();
- pendingDeleteCommit = Collections.emptySet();
}
} finally {
jedisContainer.returnInstance(commands);
@@ -118,11 +116,8 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
public void put(K key, V value) {
LOG.debug("put key '{}', value '{}'", key, value);
String redisKey = encode(keySerializer.serialize(key));
- synchronized (this) {
- pendingPrepare.put(redisKey,
- encode(valueSerializer.serialize(value)));
- pendingDeletePrepare.remove(redisKey);
- }
+ pendingPrepare.put(redisKey,
+ encode(valueSerializer.serialize(Optional.of(value))));
}
@Override
@@ -130,17 +125,11 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
LOG.debug("get key '{}'", key);
String redisKey = encode(keySerializer.serialize(key));
String redisValue = null;
- boolean found = false;
- synchronized (this) {
- if (pendingPrepare.containsKey(redisKey) || pendingDeletePrepare.contains(redisKey)) {
- redisValue = pendingPrepare.get(redisKey);
- found = true;
- } else if (pendingCommit.containsKey(redisKey) || pendingDeleteCommit.contains(redisKey)) {
- redisValue = pendingCommit.get(redisKey);
- found = true;
- }
- }
- if (!found) {
+ if (pendingPrepare.containsKey(redisKey)) {
+ redisValue = pendingPrepare.get(redisKey);
+ } else if (pendingCommit.containsKey(redisKey)) {
+ redisValue = pendingCommit.get(redisKey);
+ } else {
JedisCommands commands = null;
try {
commands = jedisContainer.getInstance();
@@ -151,7 +140,7 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
}
V value = null;
if (redisValue != null) {
- value = valueSerializer.deserialize(decode(redisValue));
+ value = valueSerializer.deserialize(decode(redisValue)).orNull();
}
LOG.debug("Value for key '{}' is '{}'", key, value);
return value;
@@ -164,17 +153,16 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
}
@Override
- public void delete(K key) {
+ public V delete(K key) {
LOG.debug("delete key '{}'", key);
String redisKey = encode(keySerializer.serialize(key));
- synchronized (this) {
- pendingDeletePrepare.add(redisKey);
- pendingPrepare.remove(redisKey);
- }
+ V curr = get(key);
+ pendingPrepare.put(redisKey, tombstone);
+ return curr;
}
@Override
- public void prepareCommit(long txid){
+ public void prepareCommit(long txid) {
LOG.debug("prepareCommit txid {}", txid);
validatePrepareTxid(txid);
JedisCommands commands = null;
@@ -188,20 +176,15 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
}
}
}
- if (!pendingPrepare.isEmpty() || !pendingDeletePrepare.isEmpty()) {
+ if (!pendingPrepare.isEmpty()) {
commands.hmset(prepareNamespace, pendingPrepare);
- commands.hdel(prepareNamespace, pendingDeletePrepare.toArray(new String[pendingDeletePrepare.size()]));
} else {
LOG.debug("Nothing to save for prepareCommit, txid {}.", txid);
}
txIds.put(PREPARE_TXID_KEY, String.valueOf(txid));
commands.hmset(txidNamespace, txIds);
- synchronized (this) {
- pendingCommit = Collections.unmodifiableMap(pendingPrepare);
- pendingDeleteCommit = Collections.unmodifiableSet(pendingDeletePrepare);
- }
+ pendingCommit = Collections.unmodifiableMap(pendingPrepare);
pendingPrepare = new ConcurrentHashMap<>();
- pendingDeletePrepare = Collections.synchronizedSet(new HashSet<String>());
} finally {
jedisContainer.returnInstance(commands);
}
@@ -214,9 +197,22 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
JedisCommands commands = null;
try {
commands = jedisContainer.getInstance();
- if (!pendingCommit.isEmpty() || !pendingDeleteCommit.isEmpty()) {
- commands.hmset(namespace, pendingCommit);
- commands.hdel(namespace, pendingDeleteCommit.toArray(new String[pendingDeleteCommit.size()]));
+ if (!pendingCommit.isEmpty()) {
+ List<String> keysToDelete = new ArrayList<>();
+ Map<String, String> keysToAdd = new HashMap<>();
+ for(Map.Entry<String, String> entry: pendingCommit.entrySet()) {
+ if (tombstone.equals(entry.getValue())) {
+ keysToDelete.add(entry.getKey());
+ } else {
+ keysToAdd.put(entry.getKey(), entry.getValue());
+ }
+ }
+ if (!keysToAdd.isEmpty()) {
+ commands.hmset(namespace, keysToAdd);
+ }
+ if (!keysToDelete.isEmpty()) {
+ commands.hdel(namespace, keysToDelete.toArray(new String[0]));
+ }
} else {
LOG.debug("Nothing to save for commit, txid {}.", txid);
}
@@ -224,7 +220,6 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
commands.hmset(txidNamespace, txIds);
commands.del(prepareNamespace);
pendingCommit = Collections.emptyMap();
- pendingDeleteCommit = Collections.emptySet();
} finally {
jedisContainer.returnInstance(commands);
}
@@ -235,14 +230,12 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
JedisCommands commands = null;
try {
commands = jedisContainer.getInstance();
- if (!pendingPrepare.isEmpty() || !pendingDeletePrepare.isEmpty()) {
+ if (!pendingPrepare.isEmpty()) {
commands.hmset(namespace, pendingPrepare);
- commands.hdel(namespace, pendingDeletePrepare.toArray(new String[pendingDeletePrepare.size()]));
} else {
LOG.debug("Nothing to save for commit");
}
- pendingPrepare = new HashMap<>();
- pendingDeletePrepare = new HashSet<>();
+ pendingPrepare = new ConcurrentHashMap<>();
} finally {
jedisContainer.returnInstance(commands);
}
@@ -270,9 +263,7 @@ public class RedisKeyValueState<K, V> implements KeyValueState<K, V> {
commands.hmset(txidNamespace, txIds);
}
pendingCommit = Collections.emptyMap();
- pendingDeleteCommit = Collections.emptySet();
pendingPrepare = new ConcurrentHashMap<>();
- pendingDeletePrepare = Collections.synchronizedSet(new HashSet<String>());
} finally {
jedisContainer.returnInstance(commands);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/0d01d974/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
----------------------------------------------------------------------
diff --git a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
index f5525d0..1bb72a7 100644
--- a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
+++ b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java
@@ -17,6 +17,7 @@
*/
package org.apache.storm.redis.state;
+import com.google.common.base.Optional;
import org.apache.storm.state.DefaultStateSerializer;
import org.apache.storm.redis.common.container.JedisCommandsInstanceContainer;
import org.junit.Before;
@@ -104,7 +105,7 @@ public class RedisKeyValueStateTest {
});
keyValueState = new RedisKeyValueState<String, String>("test", mockContainer, new DefaultStateSerializer<String>(),
- new DefaultStateSerializer<String>());
+ new DefaultStateSerializer<Optional<String>>());
}
@@ -124,7 +125,7 @@ public class RedisKeyValueStateTest {
assertEquals("1", keyValueState.get("a"));
assertEquals("2", keyValueState.get("b"));
assertEquals(null, keyValueState.get("c"));
- keyValueState.delete("a");
+ assertEquals("1", keyValueState.delete("a"));
assertEquals(null, keyValueState.get("a"));
assertEquals("2", keyValueState.get("b"));
assertEquals(null, keyValueState.get("c"));
@@ -148,8 +149,8 @@ public class RedisKeyValueStateTest {
keyValueState.rollback();
assertArrayEquals(new String[]{"1", "2", null}, getValues());
keyValueState.put("c", "3");
- keyValueState.delete("b");
- keyValueState.delete("c");
+ assertEquals("2", keyValueState.delete("b"));
+ assertEquals("3", keyValueState.delete("c"));
assertArrayEquals(new String[]{"1", null, null}, getValues());
keyValueState.prepareCommit(2);
assertArrayEquals(new String[]{"1", null, null}, getValues());
http://git-wip-us.apache.org/repos/asf/storm/blob/0d01d974/storm-core/src/jvm/org/apache/storm/state/DefaultStateSerializer.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/state/DefaultStateSerializer.java b/storm-core/src/jvm/org/apache/storm/state/DefaultStateSerializer.java
index 55e934e..d37101d 100644
--- a/storm-core/src/jvm/org/apache/storm/state/DefaultStateSerializer.java
+++ b/storm-core/src/jvm/org/apache/storm/state/DefaultStateSerializer.java
@@ -20,6 +20,7 @@ package org.apache.storm.state;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
+import org.objenesis.strategy.StdInstantiatorStrategy;
import java.util.Collections;
import java.util.List;
@@ -40,6 +41,7 @@ public class DefaultStateSerializer<T> implements Serializer<T> {
*/
public DefaultStateSerializer(List<Class<?>> classesToRegister) {
kryo = new Kryo();
+ kryo.setInstantiatorStrategy(new Kryo.DefaultInstantiatorStrategy(new StdInstantiatorStrategy()));
output = new Output(2000, 2000000000);
for (Class<?> klazz : classesToRegister) {
kryo.register(klazz);
http://git-wip-us.apache.org/repos/asf/storm/blob/0d01d974/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java b/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java
index 4774d72..72c1523 100644
--- a/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java
+++ b/storm-core/src/jvm/org/apache/storm/state/InMemoryKeyValueState.java
@@ -68,8 +68,8 @@ public class InMemoryKeyValueState<K, V> implements KeyValueState<K, V> {
}
@Override
- public void delete(K key) {
- state.remove(key);
+ public V delete(K key) {
+ return state.remove(key);
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/0d01d974/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java b/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java
index 0e1facb..7086331 100644
--- a/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java
+++ b/storm-core/src/jvm/org/apache/storm/state/KeyValueState.java
@@ -51,5 +51,5 @@ public interface KeyValueState<K, V> extends State {
*
* @param key the key
*/
- void delete(K key);
+ V delete(K key);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/0d01d974/storm-core/test/jvm/org/apache/storm/state/InMemoryKeyValueStateTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/state/InMemoryKeyValueStateTest.java b/storm-core/test/jvm/org/apache/storm/state/InMemoryKeyValueStateTest.java
index 361865b..d215627 100644
--- a/storm-core/test/jvm/org/apache/storm/state/InMemoryKeyValueStateTest.java
+++ b/storm-core/test/jvm/org/apache/storm/state/InMemoryKeyValueStateTest.java
@@ -44,6 +44,19 @@ public class InMemoryKeyValueStateTest {
}
@Test
+ public void testPutAndDelete() throws Exception {
+ keyValueState.put("a", "1");
+ keyValueState.put("b", "2");
+ assertEquals("1", keyValueState.get("a"));
+ assertEquals("2", keyValueState.get("b"));
+ assertEquals(null, keyValueState.get("c"));
+ assertEquals("1", keyValueState.delete("a"));
+ assertEquals(null, keyValueState.get("a"));
+ assertEquals("2", keyValueState.get("b"));
+ assertEquals(null, keyValueState.get("c"));
+ }
+
+ @Test
public void testPrepareCommitRollback() throws Exception {
keyValueState.put("a", "1");
keyValueState.put("b", "2");
@@ -60,6 +73,20 @@ public class InMemoryKeyValueStateTest {
assertArrayEquals(new String[]{"1", "2", "3"}, getValues());
keyValueState.rollback();
assertArrayEquals(new String[]{"1", "2", null}, getValues());
+ keyValueState.put("c", "3");
+ assertEquals("2", keyValueState.delete("b"));
+ assertEquals("3", keyValueState.delete("c"));
+ assertArrayEquals(new String[]{"1", null, null}, getValues());
+ keyValueState.prepareCommit(2);
+ assertArrayEquals(new String[]{"1", null, null}, getValues());
+ keyValueState.commit(2);
+ assertArrayEquals(new String[]{"1", null, null}, getValues());
+ keyValueState.put("b", "2");
+ keyValueState.prepareCommit(3);
+ keyValueState.put("c", "3");
+ assertArrayEquals(new String[]{"1", "2", "3"}, getValues());
+ keyValueState.rollback();
+ assertArrayEquals(new String[]{"1", null, null}, getValues());
}
private String[] getValues() {