[STORM-1917] Storm Redis TTL Control

external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisStoreBolt.java

@@ -115,6 +115,12 @@ public void execute(Tuple input) {
                     throw new IllegalArgumentException("Cannot process such data type: " + dataType);
             }
 
+            long ttlMS = storeMapper.getTTLFromTuple(input);
+
+            if (ttlMS >= 0) {
+                jedisCommand.pexpire(key, ttlMS);
+            }
+
             collector.ack(input);
         } catch (Exception e) {
             this.collector.reportError(e);

external/storm-redis/src/main/java/org/apache/storm/redis/common/mapper/RedisStoreMapper.java

@@ -17,8 +17,17 @@
  */
 package org.apache.storm.redis.common.mapper;
 
+import org.apache.storm.tuple.ITuple;
+
 /**
  * RedisStoreMapper is for defining spec. which is used for storing value to Redis.
  */
 public interface RedisStoreMapper extends TupleMapper, RedisMapper {
+    /**
+     * Extracts time to live from tuple.
+     *
+     * @param tuple
+     * @return time to live MS
+     */
+    public long getTTLFromTuple(ITuple tuple);
 }

external/storm-redis/src/test/java/org/apache/storm/redis/topology/PersistentWordCount.java

@@ -112,5 +112,10 @@ public String getKeyFromTuple(ITuple tuple) {
         public String getValueFromTuple(ITuple tuple) {
             return tuple.getStringByField("count");
         }
+
+        @Override
+        public long getTTLFromTuple(ITuple tuple) {
+            return -1;
+        }
     }
 }

external/storm-redis/src/test/java/org/apache/storm/redis/trident/WordCountStoreMapper.java

@@ -36,4 +36,9 @@ public String getKeyFromTuple(ITuple tuple) {
     public String getValueFromTuple(ITuple tuple) {
         return tuple.getInteger(1).toString();
     }
+
+    @Override
+    public long getTTLFromTuple(ITuple tuple) {
+        return -1;
+    }
 }