storm trident使用RedisState报错

错误日志如下:
2016-01-23T14:21:05.527+0800 STDIO [INFO] hkey===============================>ad_tridentpv
2016-01-23T14:21:05.565+0800 STDIO [INFO] keyValueSize================>0
2016-01-23T14:21:05.566+0800 STDIO [INFO] state========================key================>2016/01/22/11:26:00_http://twitter.com
2016-01-23T14:21:05.573+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.<init>(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
使用了RedisState保存状态,第一次提交topoloy报错,查看日志发现在multiget方法中Map<byte[], byte[]> keyValue = hgetAll(this.options.hkey.getBytes());返回的是空,导致return deserializeValues(keys, values);报空指针错误

hello_world - 大数据工程师@人民网

赞同来自: wangxiaolei

由于RedisState的multiGet第一次从redis获取state数据的时候返回值是空,需要对空值进行判断如果为空直接返回空不再创建字符串对象

要回复问题请先登录注册