kstyrc/trident-redis

Trident state implementation for Redis

trident-redis's Issues

Clojars release?

Any chance you could push a release to Clojars or some Maven repo? I'm trying to build this but I'm unfamiliar with gradle and I'm getting an error:

15:44:22.505 [ERROR] [org.gradle.BuildExceptionReporter] FAILURE: Build failed with an exception.
15:44:22.505 [ERROR] [org.gradle.BuildExceptionReporter]
15:44:22.506 [ERROR] [org.gradle.BuildExceptionReporter] * What went wrong:
15:44:22.506 [ERROR] [org.gradle.BuildExceptionReporter] Execution failed for task ':install'.
15:44:22.506 [ERROR] [org.gradle.BuildExceptionReporter] > Could not publish configuration ':archives'.
15:44:22.506 [ERROR] [org.gradle.BuildExceptionReporter] > For artifact {:trident-redis:1.0:jar}: The groupId cannot be empty.
15:44:22.510 [ERROR] [org.gradle.BuildExceptionReporter]
15:44:22.511 [ERROR] [org.gradle.BuildExceptionReporter] * Exception is:
15:44:22.512 [ERROR] [org.gradle.BuildExceptionReporter] org.gradle.api.tasks.TaskExecutionException: Execution failed for task ':install'.

Upload new version to clojars

It would be really nice if you could upload a new version to clojars with the new hash key implementation you did.

did you run it ?!

multiGet try catch should like multiPut

or it will not pool.returnResource(jedis)

storm use redisstate problem

error log:
2016-01-23T14:12:05.536+0800 b.s.util [ERROR] Async loop died!
java.lang.RuntimeException: java.lang.NullPointerException
at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:135) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:106) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:80) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.daemon.executor$fn__3439$fn__3451$fn__3498.invoke(executor.clj:755) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.util$async_loop$fn__460.invoke(util.clj:463) ~[storm-core-0.9.6.jar:0.9.6]
at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
at java.lang.Thread.run(Thread.java:662) [na:1.6.0_34]
Caused by: java.lang.NullPointerException: null
at java.lang.String.(String.java:593) ~[na:1.6.0_34]
at com.peopleyuqing.cloud.redis.RedisState.buildValuesFromMap(RedisState.java:214) ~[storm_trident_pv.jar:na]
at com.peopleyuqing.cloud.redis.RedisState.multiGet(RedisState.java:200) ~[storm_trident_pv.jar:na]
at storm.trident.state.map.CachedMap.multiGet(CachedMap.java:52) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.CachedBatchReadsMap.multiGet(CachedBatchReadsMap.java:52) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.TransactionalMap.multiUpdate(TransactionalMap.java:56) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.SnapshottableMap.multiUpdate(SnapshottableMap.java:42) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.MapCombinerAggStateUpdater.updateState(MapCombinerAggStateUpdater.java:64) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.MapCombinerAggStateUpdater.updateState(MapCombinerAggStateUpdater.java:35) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.planner.processor.PartitionPersistProcessor.finishBatch(PartitionPersistProcessor.java:98) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.planner.SubtopologyBolt.finishBatch(SubtopologyBolt.java:152) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.topology.TridentBoltExecutor.finishBatch(TridentBoltExecutor.java:252) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.topology.TridentBoltExecutor.checkFinish(TridentBoltExecutor.java:285) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.topology.TridentBoltExecutor.execute(TridentBoltExecutor.java:359) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.daemon.executor$fn__3439$tuple_action_fn__3441.invoke(executor.clj:640) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.daemon.executor$mk_task_receiver$fn__3362.invoke(executor.clj:407) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.disruptor$clojure_handler$reify__1445.onEvent(disruptor.clj:58) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:132) ~[storm-core-0.9.6.jar:0.9.6]
... 6 common frames omitted
2016-01-23T14:12:05.537+0800 b.s.d.executor [ERROR]
java.lang.RuntimeException: java.lang.NullPointerException
at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:135) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:106) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:80) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.daemon.executor$fn__3439$fn__3451$fn__3498.invoke(executor.clj:755) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.util$async_loop$fn__460.invoke(util.clj:463) ~[storm-core-0.9.6.jar:0.9.6]
at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
at java.lang.Thread.run(Thread.java:662) [na:1.6.0_34]
Caused by: java.lang.NullPointerException: null
at java.lang.String.(String.java:593) ~[na:1.6.0_34]
at com.peopleyuqing.cloud.redis.RedisState.buildValuesFromMap(RedisState.java:214) ~[storm_trident_pv.jar:na]
at com.peopleyuqing.cloud.redis.RedisState.multiGet(RedisState.java:200) ~[storm_trident_pv.jar:na]
at storm.trident.state.map.CachedMap.multiGet(CachedMap.java:52) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.CachedBatchReadsMap.multiGet(CachedBatchReadsMap.java:52) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.TransactionalMap.multiUpdate(TransactionalMap.java:56) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.SnapshottableMap.multiUpdate(SnapshottableMap.java:42) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.MapCombinerAggStateUpdater.updateState(MapCombinerAggStateUpdater.java:64) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.state.map.MapCombinerAggStateUpdater.updateState(MapCombinerAggStateUpdater.java:35) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.planner.processor.PartitionPersistProcessor.finishBatch(PartitionPersistProcessor.java:98) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.planner.SubtopologyBolt.finishBatch(SubtopologyBolt.java:152) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.topology.TridentBoltExecutor.finishBatch(TridentBoltExecutor.java:252) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.topology.TridentBoltExecutor.checkFinish(TridentBoltExecutor.java:285) ~[storm-core-0.9.6.jar:0.9.6]
at storm.trident.topology.TridentBoltExecutor.execute(TridentBoltExecutor.java:359) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.daemon.executor$fn__3439$tuple_action_fn__3441.invoke(executor.clj:640) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.daemon.executor$mk_task_receiver$fn__3362.invoke(executor.clj:407) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.disruptor$clojure_handler$reify__1445.onEvent(disruptor.clj:58) ~[storm-core-0.9.6.jar:0.9.6]
at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:132) ~[storm-core-0.9.6.jar:0.9.6]
... 6 common frames omitted

I find multiGet method
public List multiGet(List<List> keys) {
if (keys.size() == 0) {
return Collections.emptyList();
}

  if (Strings.isNullOrEmpty(this.options.hkey)) {
     String[] stringKeys = buildKeys(keys);
     List<String> values = mget(stringKeys);
     return deserializeValues(keys, values);
  } else {
     System.out.println("hkey===============================>" + this.options.hkey);
     Map<byte[], byte[]> keyValue = hgetAll(this.options.hkey.getBytes());

----------------------------------------keyValue.size() is 0 ----------------------------------

     System.out.println("keyValueSize================>"+keyValue.size());
     List<String> values = buildValuesFromMap(keys, keyValue);
     return deserializeValues(keys, values);
  }

}

Fault-Tolerant?

I'm wondering if trident-redis is fault-tolerant?

Upgrade storm dependency version

Any chance you can upgrade to a newer version of storm i.e. 0.9.0 ?

CacheMap class doesnt take keyFactory input

Shouldn't the cacheMap should also take keyFactory?

public CachedMap(IBackingMap<T> delegate, int cacheSize) {
    _cache = new LRUMap<List<Object>, T>(cacheSize);
    _delegate = delegate;
}

It will cause issue at this step

 if(_cache.containsKey(key))  {
            results.put(key, _cache.get(key));
 } 

Since the key cached will be different from the one stored in redis.