// Path: storm/src/main/java/com/fksm/utils/CuratorLockAcquire.java
// public class CuratorLockAcquire  {
//     private static final Logger LOGGER = LoggerFactory.getLogger(CuratorLockAcquire.class);
// 
//     private CuratorFramework client = null;
// 
//     public CuratorLockAcquire() {}
// 
//     public synchronized CuratorFramework getCuratorClient() {
//         if (client == null) {
//             try {
//                 String zkHostPort = ConfigPropertieUtil.getInstance().getString("curator.lock.zk.connect");
//                 RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
//                 client = CuratorFrameworkFactory.newClient(zkHostPort, retryPolicy);
//                 client.start();
//             } catch (Exception e) {
//                 LOGGER.error("Get curatorClient error.", e);
//             }
//         }
// 
//         return client;
//     }
// }
// 
// Path: storm/src/main/java/com/fksm/utils/RedisKeysQueueManager.java
// public class RedisKeysQueueManager {
//     private static ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();
// 
//     public static void add(String key){
//         queue.add(key);
//     }
// 
//     public static String pop(){
//         if(!queue.isEmpty()) {
//             return  queue.poll();
//         }
//         return null;
//     }
// 
//     public static boolean isEmpty(){
//         return queue.isEmpty();
//     }
// 
//     public static int size(){
//         return queue.size();
//     }
// }


import com.fksm.utils.CuratorLockAcquire;
import com.fksm.utils.RedisKeysQueueManager;
import com.netflix.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
package com.fksm.storm.bolt;



/**
 * remove data from redis
 */
public class RemoveRedisBolt extends BaseRedisBasicBolt {

    private static final Logger logger = LoggerFactory.getLogger(RemoveRedisBolt.class);

// Path: storm/src/main/java/com/fksm/utils/CuratorLockAcquire.java
// public class CuratorLockAcquire  {
//     private static final Logger LOGGER = LoggerFactory.getLogger(CuratorLockAcquire.class);
// 
//     private CuratorFramework client = null;
// 
//     public CuratorLockAcquire() {}
// 
//     public synchronized CuratorFramework getCuratorClient() {
//         if (client == null) {
//             try {
//                 String zkHostPort = ConfigPropertieUtil.getInstance().getString("curator.lock.zk.connect");
//                 RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
//                 client = CuratorFrameworkFactory.newClient(zkHostPort, retryPolicy);
//                 client.start();
//             } catch (Exception e) {
//                 LOGGER.error("Get curatorClient error.", e);
//             }
//         }
// 
//         return client;
//     }
// }
// 
// Path: storm/src/main/java/com/fksm/utils/RedisKeysQueueManager.java
// public class RedisKeysQueueManager {
//     private static ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();
// 
//     public static void add(String key){
//         queue.add(key);
//     }
// 
//     public static String pop(){
//         if(!queue.isEmpty()) {
//             return  queue.poll();
//         }
//         return null;
//     }
// 
//     public static boolean isEmpty(){
//         return queue.isEmpty();
//     }
// 
//     public static int size(){
//         return queue.size();
//     }
// }


// Path: storm/src/main/java/com/fksm/storm/bolt/RemoveRedisBolt.java
import com.fksm.utils.CuratorLockAcquire;
import com.fksm.utils.RedisKeysQueueManager;
import com.netflix.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

package com.fksm.storm.bolt;



/**
 * remove data from redis
 */
public class RemoveRedisBolt extends BaseRedisBasicBolt {

    private static final Logger logger = LoggerFactory.getLogger(RemoveRedisBolt.class);

    private CuratorLockAcquire curatorLockAcquire;