// Path: storm/src/main/java/com/fksm/storm/hdfs/action/MoveStormToLogAction.java
// public class MoveStormToLogAction implements RotationAction {
//     private static final Logger logger = LoggerFactory.getLogger(MoveStormToLogAction.class);
// 
//     private String destination;
// 
//     public MoveStormToLogAction withDestination(String destDir){
//         destination = destDir;
//         return this;
//     }
// 
//     @Override
//     public void execute(FileSystem fileSystem, Path filePath) throws IOException {
//         Path destPath = new Path(destination, filePath.getName());
//         logger.info("Moving file {} to {}", filePath, destPath);
//         boolean success = fileSystem.rename(filePath, destPath);
//         return;
//     }
// }
// 
// Path: storm/src/main/java/com/fksm/utils/ConfigPropertieUtil.java
// public class ConfigPropertieUtil {
// 
//     private static final Logger LOGGER = LoggerFactory.getLogger(ConfigPropertieUtil.class);
// 
//     private static Config config;
// 
//     static {
//         config = ConfigFactory.load();
//     }
// 
//     public static Config getInstance(){
//         return config;
//     }
// 
//     public static Config getInstance(String configFile){
//         config = ConfigFactory.load(configFile);
//         config.checkValid(ConfigFactory.defaultReference(), "simple-lib");
//         return config;
//     }
// }


import com.fksm.storm.bolt.*;
import com.fksm.storm.hdfs.action.MoveStormToLogAction;
import com.fksm.utils.ConfigPropertieUtil;
import org.apache.hadoop.io.SequenceFile;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.hdfs.bolt.HdfsBolt;
import org.apache.storm.hdfs.bolt.SequenceFileBolt;
import org.apache.storm.hdfs.bolt.format.*;
import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy;
import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy.*;
import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;
import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
import org.apache.storm.kafka.*;
import org.apache.storm.spout.SchemeAsMultiScheme;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseWindowedBolt.*;
import org.apache.storm.tuple.Fields;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
            conf.setMaxTaskParallelism(numWorkers);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(name, conf, builder.createTopology());
            Thread.sleep(60000);
            cluster.shutdown();
        }
    }


    private static HdfsBolt buildHdfsBolt(String hdfsUrl,String prefix, Fields fields){
        // use "|" instead of "," for field delimiter
        RecordFormat format = new DelimitedRecordFormat()
                .withFieldDelimiter(" : ").withFields(fields);

        // sync the filesystem after every 1k tuples
        SyncPolicy syncPolicy = new CountSyncPolicy(1000);

        // rotate files
        FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, Units.MB);

        FileNameFormat fileNameFormat = new DefaultFileNameFormat()
                .withPath("/storm/").withPrefix(prefix).withExtension(".seq");

        HdfsBolt hdfsBolt = new HdfsBolt()
                .withFsUrl(hdfsUrl)
                .withFileNameFormat(fileNameFormat)
                .withRecordFormat(format)
                .withRotationPolicy(rotationPolicy)
                .withSyncPolicy(syncPolicy)
                .withRetryCount(5)
// Path: storm/src/main/java/com/fksm/storm/hdfs/action/MoveStormToLogAction.java
// public class MoveStormToLogAction implements RotationAction {
//     private static final Logger logger = LoggerFactory.getLogger(MoveStormToLogAction.class);
// 
//     private String destination;
// 
//     public MoveStormToLogAction withDestination(String destDir){
//         destination = destDir;
//         return this;
//     }
// 
//     @Override
//     public void execute(FileSystem fileSystem, Path filePath) throws IOException {
//         Path destPath = new Path(destination, filePath.getName());
//         logger.info("Moving file {} to {}", filePath, destPath);
//         boolean success = fileSystem.rename(filePath, destPath);
//         return;
//     }
// }
// 
// Path: storm/src/main/java/com/fksm/utils/ConfigPropertieUtil.java
// public class ConfigPropertieUtil {
// 
//     private static final Logger LOGGER = LoggerFactory.getLogger(ConfigPropertieUtil.class);
// 
//     private static Config config;
// 
//     static {
//         config = ConfigFactory.load();
//     }
// 
//     public static Config getInstance(){
//         return config;
//     }
// 
//     public static Config getInstance(String configFile){
//         config = ConfigFactory.load(configFile);
//         config.checkValid(ConfigFactory.defaultReference(), "simple-lib");
//         return config;
//     }
// }


// Path: storm/src/main/java/com/fksm/storm/topology/KafkaTopology.java
import com.fksm.storm.bolt.*;
import com.fksm.storm.hdfs.action.MoveStormToLogAction;
import com.fksm.utils.ConfigPropertieUtil;
import org.apache.hadoop.io.SequenceFile;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.hdfs.bolt.HdfsBolt;
import org.apache.storm.hdfs.bolt.SequenceFileBolt;
import org.apache.storm.hdfs.bolt.format.*;
import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy;
import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy.*;
import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;
import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
import org.apache.storm.kafka.*;
import org.apache.storm.spout.SchemeAsMultiScheme;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseWindowedBolt.*;
import org.apache.storm.tuple.Fields;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;

            conf.setMaxTaskParallelism(numWorkers);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(name, conf, builder.createTopology());
            Thread.sleep(60000);
            cluster.shutdown();
        }
    }


    private static HdfsBolt buildHdfsBolt(String hdfsUrl,String prefix, Fields fields){
        // use "|" instead of "," for field delimiter
        RecordFormat format = new DelimitedRecordFormat()
                .withFieldDelimiter(" : ").withFields(fields);

        // sync the filesystem after every 1k tuples
        SyncPolicy syncPolicy = new CountSyncPolicy(1000);

        // rotate files
        FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, Units.MB);

        FileNameFormat fileNameFormat = new DefaultFileNameFormat()
                .withPath("/storm/").withPrefix(prefix).withExtension(".seq");

        HdfsBolt hdfsBolt = new HdfsBolt()
                .withFsUrl(hdfsUrl)
                .withFileNameFormat(fileNameFormat)
                .withRecordFormat(format)
                .withRotationPolicy(rotationPolicy)
                .withSyncPolicy(syncPolicy)
                .withRetryCount(5)
                .addRotationAction(new MoveStormToLogAction().withDestination("/log"));