[32/51] [partial] incubator-metron git commit: METRON-113 Project Reorganization (merrimanr) closes apache/incubator-metron#88

[rm...@apache.org]

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/SimpleEnrichmentFlatFileLoader.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/SimpleEnrichmentFlatFileLoader.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/SimpleEnrichmentFlatFileLoader.java
new file mode 100644
index 0000000..cbd3beb
--- /dev/null
+++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/flatfile/SimpleEnrichmentFlatFileLoader.java
@@ -0,0 +1,261 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.dataloads.nonbulk.flatfile;
+
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableList;
+import org.apache.commons.cli.*;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.util.GenericOptionsParser;
+import org.apache.log4j.PropertyConfigurator;
+import org.apache.metron.dataloads.extractor.Extractor;
+import org.apache.metron.dataloads.extractor.ExtractorHandler;
+import org.apache.metron.dataloads.extractor.inputformat.WholeFileFormat;
+import org.apache.metron.common.configuration.EnrichmentConfig;
+import org.apache.metron.hbase.HTableProvider;
+import org.apache.metron.enrichment.converter.HbaseConverter;
+import org.apache.metron.enrichment.converter.EnrichmentConverter;
+import org.apache.metron.enrichment.lookup.LookupKV;
+import org.apache.metron.common.utils.JSONUtils;
+
+import javax.annotation.Nullable;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Stack;
+
+public class SimpleEnrichmentFlatFileLoader {
+  private static abstract class OptionHandler implements Function<String, Option> {}
+  public static enum LoadOptions {
+    HELP("h", new OptionHandler() {
+
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        return new Option(s, "help", false, "Generate Help screen");
+      }
+    })
+    ,HBASE_TABLE("t", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "hbase_table", true, "HBase table to ingest the data into.");
+        o.setArgName("TABLE");
+        o.setRequired(true);
+        return o;
+      }
+    })
+    ,HBASE_CF("c", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "hbase_cf", true, "HBase column family to ingest the data into.");
+        o.setArgName("CF");
+        o.setRequired(true);
+        return o;
+      }
+    })
+    ,EXTRACTOR_CONFIG("e", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "extractor_config", true, "JSON Document describing the extractor for this input data source");
+        o.setArgName("JSON_FILE");
+        o.setRequired(true);
+        return o;
+      }
+    })
+    ,ENRICHMENT_CONFIG("n", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "enrichment_config", true
+                , "JSON Document describing the enrichment configuration details." +
+                "  This is used to associate an enrichment type with a field type in zookeeper."
+        );
+        o.setArgName("JSON_FILE");
+        o.setRequired(false);
+        return o;
+      }
+    })
+    ,LOG4J_PROPERTIES("l", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "log4j", true, "The log4j properties file to load");
+        o.setArgName("FILE");
+        o.setRequired(false);
+        return o;
+      }
+    })
+    ,INPUT("i", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "input", true, "The CSV File to load");
+        o.setArgName("FILE");
+        o.setRequired(true);
+        return o;
+      }
+    })
+    ;
+    Option option;
+    String shortCode;
+    LoadOptions(String shortCode, OptionHandler optionHandler) {
+      this.shortCode = shortCode;
+      this.option = optionHandler.apply(shortCode);
+    }
+
+    public boolean has(CommandLine cli) {
+      return cli.hasOption(shortCode);
+    }
+
+    public String get(CommandLine cli) {
+      return cli.getOptionValue(shortCode);
+    }
+
+    public static CommandLine parse(CommandLineParser parser, String[] args) {
+      try {
+        CommandLine cli = parser.parse(getOptions(), args);
+        if(HELP.has(cli)) {
+          printHelp();
+          System.exit(0);
+        }
+        return cli;
+      } catch (ParseException e) {
+        System.err.println("Unable to parse args: " + Joiner.on(' ').join(args));
+        e.printStackTrace(System.err);
+        printHelp();
+        System.exit(-1);
+        return null;
+      }
+    }
+
+    public static void printHelp() {
+      HelpFormatter formatter = new HelpFormatter();
+      formatter.printHelp( "SimpleEnrichmentFlatFileLoader", getOptions());
+    }
+
+    public static Options getOptions() {
+      Options ret = new Options();
+      for(LoadOptions o : LoadOptions.values()) {
+        ret.addOption(o.option);
+      }
+      return ret;
+    }
+  }
+  public static List<File> getFiles(File root) {
+    if(!root.isDirectory())  {
+      return ImmutableList.of(root);
+    }
+    List<File> ret = new ArrayList<>();
+    Stack<File> stack = new Stack<File>();
+    stack.push(root);
+    while(!stack.isEmpty()) {
+      File f = stack.pop();
+      if(f.isDirectory()) {
+        for(File child : f.listFiles()) {
+          stack.push(child);
+        }
+      }
+      else {
+        ret.add(f);
+      }
+    }
+    return ret;
+  }
+
+  public HTableProvider getProvider() {
+    return new HTableProvider();
+  }
+
+  public List<Put> extract( String line
+                     , Extractor extractor
+                     , String cf
+                     , HbaseConverter converter
+                     ) throws IOException
+  {
+    List<Put> ret = new ArrayList<>();
+    Iterable<LookupKV> kvs = extractor.extract(line);
+    for(LookupKV kv : kvs) {
+      Put put = converter.toPut(cf, kv.getKey(), kv.getValue());
+      ret.add(put);
+    }
+    return ret;
+  }
+
+
+  public void loadFile( File inputFile
+                      , Extractor extractor
+                      , HTableInterface table
+                      , String cf
+                      , HbaseConverter converter
+                      , boolean lineByLine
+                      ) throws IOException
+  {
+    if(!lineByLine) {
+      table.put(extract(FileUtils.readFileToString(inputFile), extractor, cf, converter));
+    }
+    else {
+      BufferedReader br = new BufferedReader(new FileReader(inputFile));
+      for(String line = null;(line = br.readLine()) != null;) {
+        table.put(extract(line, extractor, cf, converter));
+      }
+    }
+  }
+  public static void main(String... argv) throws Exception {
+    Configuration conf = HBaseConfiguration.create();
+    String[] otherArgs = new GenericOptionsParser(conf, argv).getRemainingArgs();
+
+    CommandLine cli = LoadOptions.parse(new PosixParser(), otherArgs);
+    if(LoadOptions.LOG4J_PROPERTIES.has(cli)) {
+      PropertyConfigurator.configure(LoadOptions.LOG4J_PROPERTIES.get(cli));
+    }
+    ExtractorHandler handler = ExtractorHandler.load(
+            FileUtils.readFileToString(new File(LoadOptions.EXTRACTOR_CONFIG.get(cli)))
+    );
+    boolean lineByLine = !handler.getInputFormatHandler().getClass().equals(WholeFileFormat.class);
+    Extractor e = handler.getExtractor();
+    EnrichmentConfig enrichmentConfig = null;
+    if(LoadOptions.ENRICHMENT_CONFIG.has(cli)) {
+      enrichmentConfig = JSONUtils.INSTANCE.load( new File(LoadOptions.ENRICHMENT_CONFIG.get(cli))
+              , EnrichmentConfig.class
+      );
+    }
+    HbaseConverter converter = new EnrichmentConverter();
+    List<File> inputFiles = getFiles(new File(LoadOptions.INPUT.get(cli)));
+    SimpleEnrichmentFlatFileLoader loader = new SimpleEnrichmentFlatFileLoader();
+    HTableInterface table = loader.getProvider()
+            .getTable(conf, LoadOptions.HBASE_TABLE.get(cli));
+
+    for (File f : inputFiles) {
+      loader.loadFile(f, e, table, LoadOptions.HBASE_CF.get(cli), converter, lineByLine);
+    }
+    if(enrichmentConfig != null) {
+      enrichmentConfig.updateSensorConfigs();
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/ConnectionType.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/ConnectionType.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/ConnectionType.java
new file mode 100644
index 0000000..77d1698
--- /dev/null
+++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/ConnectionType.java
@@ -0,0 +1,23 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metron.dataloads.nonbulk.taxii;
+
+public enum ConnectionType {
+   POLL, DISCOVER;
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TableInfo.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TableInfo.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TableInfo.java
new file mode 100644
index 0000000..6bbf8e3
--- /dev/null
+++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TableInfo.java
@@ -0,0 +1,71 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metron.dataloads.nonbulk.taxii;
+
+import com.google.common.base.Splitter;
+import com.google.common.collect.Iterables;
+
+public class TableInfo {
+    private String tableName;
+    private String columnFamily;
+    public TableInfo(String s) {
+        Iterable<String> i = Splitter.on(":").split(s);
+        if(Iterables.size(i) != 2) {
+            throw new IllegalStateException("Malformed table:cf => " + s);
+        }
+        tableName = Iterables.getFirst(i, null);
+        columnFamily = Iterables.getLast(i);
+    }
+
+    public String getTableName() {
+        return tableName;
+    }
+
+    public String getColumnFamily() {
+        return columnFamily;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        TableInfo tableInfo = (TableInfo) o;
+
+        if (getTableName() != null ? !getTableName().equals(tableInfo.getTableName()) : tableInfo.getTableName() != null)
+            return false;
+        return getColumnFamily() != null ? getColumnFamily().equals(tableInfo.getColumnFamily()) : tableInfo.getColumnFamily() == null;
+
+    }
+
+    @Override
+    public int hashCode() {
+        int result = getTableName() != null ? getTableName().hashCode() : 0;
+        result = 31 * result + (getColumnFamily() != null ? getColumnFamily().hashCode() : 0);
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        return "TableInfo{" +
+                "tableName='" + tableName + '\'' +
+                ", columnFamily='" + columnFamily + '\'' +
+                '}';
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiConnectionConfig.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiConnectionConfig.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiConnectionConfig.java
new file mode 100644
index 0000000..678f98b
--- /dev/null
+++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiConnectionConfig.java
@@ -0,0 +1,222 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metron.dataloads.nonbulk.taxii;
+
+import com.google.common.base.Joiner;
+import org.apache.metron.dataloads.extractor.stix.types.ObjectTypeHandlers;
+import org.codehaus.jackson.map.ObjectMapper;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.nio.charset.Charset;
+import java.text.DateFormat;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+public class TaxiiConnectionConfig {
+  final static ObjectMapper _mapper = new ObjectMapper();
+  private URL endpoint;
+  private int port = 443;
+  private URL proxy;
+  private String username;
+  private String password;
+  private ConnectionType type;
+  private String collection = "default";
+  private String subscriptionId = null;
+  private Date beginTime;
+  private String table;
+  private String columnFamily;
+  private Set<String> allowedIndicatorTypes = new HashSet<String>();
+
+  public TaxiiConnectionConfig withAllowedIndicatorTypes(List<String> indicatorTypes) {
+    allowedIndicatorTypes = new HashSet(indicatorTypes);
+    return this;
+  }
+
+  public TaxiiConnectionConfig withTable(String table) {
+    this.table = table;
+    return this;
+  }
+  public TaxiiConnectionConfig withColumnFamily(String cf) {
+    this.columnFamily = cf;
+    return this;
+  }
+  public TaxiiConnectionConfig withBeginTime(Date time) {
+    this.beginTime = time;
+    return this;
+  }
+  public TaxiiConnectionConfig withSubscriptionId(String subId) {
+    this.subscriptionId = subId;
+    return this;
+  }
+  public TaxiiConnectionConfig withCollection(String collection) {
+    this.collection = collection;
+    return this;
+  }
+
+  public TaxiiConnectionConfig withPort(int port) {
+    this.port = port;
+    return this;
+  }
+  public TaxiiConnectionConfig withEndpoint(URL endpoint) {
+    this.endpoint = endpoint;
+    return this;
+  }
+  public TaxiiConnectionConfig withProxy(URL proxy) {
+    this.proxy = proxy;
+    return this;
+  }
+  public TaxiiConnectionConfig withUsername(String username) {
+    this.username = username;
+    return this;
+  }
+  public TaxiiConnectionConfig withPassword(String password) {
+    this.password = password;
+    return this;
+  }
+  public TaxiiConnectionConfig withConnectionType(ConnectionType type) {
+    this.type= type;
+    return this;
+  }
+
+  public void setEndpoint(String endpoint) throws MalformedURLException {
+    this.endpoint = new URL(endpoint);
+  }
+
+  public void setPort(int port) {
+    this.port = port;
+  }
+
+  public void setProxy(String proxy) throws MalformedURLException {
+    this.proxy = new URL(proxy);
+  }
+
+  public void setUsername(String username) {
+    this.username = username;
+  }
+
+  public void setPassword(String password) {
+    this.password = password;
+  }
+
+  public void setType(ConnectionType type) {
+    this.type = type;
+  }
+
+  public void setCollection(String collection) {
+    this.collection = collection;
+  }
+
+  public void setSubscriptionId(String subscriptionId) {
+    this.subscriptionId = subscriptionId;
+  }
+
+  public void setBeginTime(String beginTime) throws ParseException {
+    SimpleDateFormat sdf = (SimpleDateFormat)DateFormat.getDateInstance(DateFormat.MEDIUM);
+    this.beginTime = sdf.parse(beginTime);
+  }
+
+  public String getTable() {
+    return table;
+  }
+
+  public void setTable(String table) {
+    this.table = table;
+  }
+
+  public String getColumnFamily() {
+    return columnFamily;
+  }
+
+  public void setColumnFamily(String columnFamily) {
+    this.columnFamily = columnFamily;
+  }
+
+  public Date getBeginTime() {
+    return beginTime;
+  }
+  public int getPort() {
+    return port;
+  }
+  public URL getEndpoint() {
+    return endpoint;
+  }
+
+  public URL getProxy() {
+    return proxy;
+  }
+
+  public String getUsername() {
+    return username;
+  }
+
+  public String getPassword() {
+    return password;
+  }
+
+  public ConnectionType getType() {
+    return type;
+  }
+
+  public String getCollection() {
+    return collection;
+  }
+  public String getSubscriptionId() {
+    return subscriptionId;
+  }
+
+  public void setAllowedIndicatorTypes(List<String> allowedIndicatorTypes) {
+    withAllowedIndicatorTypes(allowedIndicatorTypes);
+  }
+
+  public Set<String> getAllowedIndicatorTypes() {
+    return allowedIndicatorTypes;
+  }
+  public static synchronized TaxiiConnectionConfig load(InputStream is) throws IOException {
+    TaxiiConnectionConfig ret = _mapper.readValue(is, TaxiiConnectionConfig.class);
+    return ret;
+  }
+  public static synchronized TaxiiConnectionConfig load(String s, Charset c) throws IOException {
+    return load( new ByteArrayInputStream(s.getBytes(c)));
+  }
+  public static synchronized TaxiiConnectionConfig load(String s) throws IOException {
+    return load( s, Charset.defaultCharset());
+  }
+
+  @Override
+  public String toString() {
+    return "TaxiiConnectionConfig{" +
+            "endpoint=" + endpoint +
+            ", port=" + port +
+            ", proxy=" + proxy +
+            ", username='" + username + '\'' +
+            ", password=" + (password == null?"null" : "'******'") +
+            ", type=" + type +
+            ", allowedIndicatorTypes=" + Joiner.on(',').join(allowedIndicatorTypes)+
+            ", collection='" + collection + '\'' +
+            ", subscriptionId='" + subscriptionId + '\'' +
+            ", beginTime=" + beginTime +
+            ", table=" + table + ":" + columnFamily+
+            '}';
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiHandler.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiHandler.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiHandler.java
new file mode 100644
index 0000000..be571e1
--- /dev/null
+++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiHandler.java
@@ -0,0 +1,406 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metron.dataloads.nonbulk.taxii;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.AuthCache;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.config.Registry;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.conn.socket.ConnectionSocketFactory;
+import org.apache.http.conn.socket.PlainConnectionSocketFactory;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.conn.ssl.SSLContextBuilder;
+import org.apache.http.conn.ssl.TrustSelfSignedStrategy;
+import org.apache.http.impl.auth.BasicScheme;
+import org.apache.http.impl.client.BasicAuthCache;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.log4j.Logger;
+import org.apache.metron.dataloads.extractor.Extractor;
+import org.apache.metron.enrichment.converter.EnrichmentConverter;
+import org.apache.metron.enrichment.converter.EnrichmentKey;
+import org.apache.metron.enrichment.converter.EnrichmentValue;
+import org.apache.metron.enrichment.lookup.LookupKV;
+import org.mitre.taxii.client.HttpClient;
+import org.mitre.taxii.messages.xml11.*;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+
+import javax.xml.bind.JAXBException;
+import javax.xml.datatype.DatatypeConfigurationException;
+import javax.xml.datatype.DatatypeFactory;
+import javax.xml.datatype.XMLGregorianCalendar;
+import javax.xml.transform.Transformer;
+import javax.xml.transform.TransformerException;
+import javax.xml.transform.TransformerFactory;
+import javax.xml.transform.dom.DOMSource;
+import javax.xml.transform.stream.StreamResult;
+import java.io.*;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+public class TaxiiHandler extends TimerTask {
+    private static final Logger LOG = Logger.getLogger(TaxiiHandler.class);
+
+    private static ThreadLocal<TaxiiXmlFactory> xmlFactory = new ThreadLocal<TaxiiXmlFactory>() {
+        @Override
+        protected TaxiiXmlFactory initialValue() {
+            return new TaxiiXmlFactory();
+        }
+    };
+    private static ThreadLocal<ObjectFactory> messageFactory = new ThreadLocal<ObjectFactory>() {
+        @Override
+        protected ObjectFactory initialValue() {
+            return new ObjectFactory();
+        }
+    };
+
+    private HttpClient taxiiClient;
+    private URL endpoint;
+    private Extractor extractor;
+    private String hbaseTable;
+    private String columnFamily;
+    private Map<String, HTableInterface> connectionCache = new HashMap<>();
+    private HttpClientContext context;
+    private String collection;
+    private String subscriptionId;
+    private EnrichmentConverter converter = new EnrichmentConverter();
+    private Date beginTime;
+    private Configuration config;
+    private boolean inProgress = false;
+    private Set<String> allowedIndicatorTypes;
+    public TaxiiHandler( TaxiiConnectionConfig connectionConfig
+                       , Extractor extractor
+                       , Configuration config
+                       ) throws Exception
+    {
+        LOG.info("Loading configuration: " + connectionConfig);
+        this.allowedIndicatorTypes = connectionConfig.getAllowedIndicatorTypes();
+        this.extractor = extractor;
+        this.collection = connectionConfig.getCollection();
+        this.subscriptionId = connectionConfig.getSubscriptionId();
+        hbaseTable = connectionConfig.getTable();
+        columnFamily = connectionConfig.getColumnFamily();
+        this.beginTime = connectionConfig.getBeginTime();
+        this.config = config;
+        initializeClient(connectionConfig);
+        LOG.info("Configured, starting polling " + endpoint + " for " + collection);
+    }
+
+    protected synchronized HTableInterface getTable(String table) throws IOException {
+        HTableInterface ret = connectionCache.get(table);
+        if(ret == null) {
+            ret = createHTable(table);
+            connectionCache.put(table, ret);
+        }
+        return ret;
+    }
+
+    protected synchronized HTableInterface createHTable(String tableInfo) throws IOException {
+        return new HTable(config, tableInfo);
+    }
+    /**
+     * The action to be performed by this timer task.
+     */
+    @Override
+    public void run() {
+        if(inProgress) {
+            return;
+        }
+        Date ts = new Date();
+        LOG.info("Polling..." + new SimpleDateFormat().format(ts));
+        try {
+            inProgress = true;
+            // Prepare the message to send.
+            String sessionID = MessageHelper.generateMessageId();
+            PollRequest request = messageFactory.get().createPollRequest()
+                    .withMessageId(sessionID)
+                    .withCollectionName(collection);
+            if (subscriptionId != null) {
+                request = request.withSubscriptionID(subscriptionId);
+            } else {
+                request = request.withPollParameters(messageFactory.get().createPollParametersType());
+            }
+            if (beginTime != null) {
+                Calendar gc = GregorianCalendar.getInstance();
+                gc.setTime(beginTime);
+                XMLGregorianCalendar gTime = null;
+                try {
+                    gTime = DatatypeFactory.newInstance().newXMLGregorianCalendar((GregorianCalendar) gc).normalize();
+                } catch (DatatypeConfigurationException e) {
+                    LOG.error("Unable to set the begin time", e);
+                }
+                gTime.setFractionalSecond(null);
+                LOG.info("Begin Time: " + gTime);
+                request.setExclusiveBeginTimestamp(gTime);
+            }
+
+            try {
+                PollResponse response = call(request, PollResponse.class);
+                LOG.info("Got Poll Response with " + response.getContentBlocks().size() + " blocks");
+                int numProcessed = 0;
+                long avgTimeMS = 0;
+                long timeStartedBlock = System.currentTimeMillis();
+                for (ContentBlock block : response.getContentBlocks()) {
+                    AnyMixedContentType content = block.getContent();
+                    for (Object o : content.getContent()) {
+                        numProcessed++;
+                        long timeS = System.currentTimeMillis();
+                        String xml = null;
+                        if (o instanceof Element) {
+                            Element element = (Element) o;
+                            xml = getStringFromDocument(element.getOwnerDocument());
+                            if(LOG.isDebugEnabled() && Math.random() < 0.01) {
+                                LOG.debug("Random Stix doc: " + xml);
+                            }
+                            for (LookupKV<EnrichmentKey, EnrichmentValue> kv : extractor.extract(xml)) {
+                                if(allowedIndicatorTypes.isEmpty()
+                                || allowedIndicatorTypes.contains(kv.getKey().type)
+                                  )
+                                {
+                                    kv.getValue().getMetadata().put("source_type", "taxii");
+                                    kv.getValue().getMetadata().put("taxii_url", endpoint.toString());
+                                    kv.getValue().getMetadata().put("taxii_collection", collection);
+                                    Put p = converter.toPut(columnFamily, kv.getKey(), kv.getValue());
+                                    HTableInterface table = getTable(hbaseTable);
+                                    table.put(p);
+                                    LOG.info("Found Threat Intel: " + kv.getKey() + " => " + kv.getValue());
+                                }
+                            }
+                        }
+                        avgTimeMS += System.currentTimeMillis() - timeS;
+                    }
+                    if( (numProcessed + 1) % 100 == 0) {
+                        LOG.info("Processed " + numProcessed + " in " + (System.currentTimeMillis() - timeStartedBlock) + " ms, avg time: " + avgTimeMS / content.getContent().size());
+                        timeStartedBlock = System.currentTimeMillis();
+                        avgTimeMS = 0;
+                        numProcessed = 0;
+                    }
+                }
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+                throw new RuntimeException("Unable to make request", e);
+            }
+        }
+        finally {
+            inProgress = false;
+            beginTime = ts;
+        }
+    }
+    public String getStringFromDocument(Document doc)
+    {
+        try
+        {
+            DOMSource domSource = new DOMSource(doc);
+            StringWriter writer = new StringWriter();
+            StreamResult result = new StreamResult(writer);
+            TransformerFactory tf = TransformerFactory.newInstance();
+            Transformer transformer = tf.newTransformer();
+            transformer.transform(domSource, result);
+            return writer.toString();
+        }
+        catch(TransformerException ex)
+        {
+            ex.printStackTrace();
+            return null;
+        }
+    }
+    private <RESPONSE_T> RESPONSE_T call( Object request, Class<RESPONSE_T> responseClazz) throws URISyntaxException, JAXBException, IOException {
+        return call(taxiiClient, endpoint.toURI(), request, context, responseClazz);
+    }
+
+    private void initializeClient(TaxiiConnectionConfig config) throws Exception {
+        LOG.info("Initializing client..");
+        if(context == null) {
+            context = createContext(config.getEndpoint(), config.getUsername(), config.getPassword(), config.getPort());
+        }
+        URL endpoint = config.getEndpoint();
+        if(config.getType() == ConnectionType.DISCOVER) {
+            LOG.info("Discovering endpoint");
+            endpoint = discoverPollingClient(config.getProxy(), endpoint, config.getUsername(), config.getPassword(), context, collection).pollEndpoint;
+            this.endpoint = endpoint;
+            LOG.info("Discovered endpoint as " + endpoint);
+        }
+        taxiiClient = buildClient(config.getProxy(), config.getUsername(), config.getPassword());
+    }
+
+    private static class DiscoveryResults {
+        URL pollEndpoint;
+        URL collectionManagementEndpoint;
+        List<String> collections = new ArrayList<>();
+    }
+    private static DiscoveryResults discoverPollingClient(URL proxy, URL endpoint, String username, String password, HttpClientContext context, String defaultCollection) throws Exception {
+
+        DiscoveryResults results = new DiscoveryResults();
+        {
+            HttpClient discoverClient = buildClient(proxy, username, password);
+            String sessionID = MessageHelper.generateMessageId();
+            // Prepare the message to send.
+            DiscoveryRequest request = messageFactory.get().createDiscoveryRequest()
+                    .withMessageId(sessionID);
+            DiscoveryResponse response = call(discoverClient, endpoint.toURI(), request, context, DiscoveryResponse.class);
+            for (ServiceInstanceType serviceInstance : response.getServiceInstances()) {
+                if (serviceInstance.isAvailable() && serviceInstance.getServiceType() == ServiceTypeEnum.POLL) {
+                    results.pollEndpoint = new URL(serviceInstance.getAddress());
+                }
+                else if(serviceInstance.isAvailable() && serviceInstance.getServiceType() == ServiceTypeEnum.COLLECTION_MANAGEMENT) {
+                    results.collectionManagementEndpoint= new URL(serviceInstance.getAddress());
+                }
+            }
+            if (results.pollEndpoint == null) {
+                throw new RuntimeException("Unable to discover a poll TAXII feed");
+            }
+        }
+        if(defaultCollection == null)
+        //get collections
+        {
+            HttpClient discoverClient = buildClient(proxy, username, password);
+            String sessionID = MessageHelper.generateMessageId();
+            CollectionInformationRequest request = messageFactory.get().createCollectionInformationRequest()
+                                                                 .withMessageId(sessionID);
+            CollectionInformationResponse response = call(discoverClient, results.collectionManagementEndpoint.toURI(), request, context, CollectionInformationResponse.class);
+            LOG.info("Unable to find the default collection; available collections are:");
+            for(CollectionRecordType c : response.getCollections()) {
+                LOG.info(c.getCollectionName());
+                results.collections.add(c.getCollectionName());
+            }
+            System.exit(0);
+        }
+        return results;
+    }
+
+    private static HttpClientContext createContext(URL endpoint, String username, String password, int port) {
+        HttpClientContext context = null;
+        HttpHost target = new HttpHost(endpoint.getHost(), port, endpoint.getProtocol());
+        if (username != null && password != null) {
+
+            CredentialsProvider credsProvider = new BasicCredentialsProvider();
+            credsProvider.setCredentials(
+                    new AuthScope(target.getHostName(), target.getPort()),
+                    new UsernamePasswordCredentials(username, password));
+
+            // http://hc.apache.org/httpcomponents-client-ga/tutorial/html/authentication.html
+            AuthCache authCache = new BasicAuthCache();
+            authCache.put(target, new BasicScheme());
+
+            // Add AuthCache to the execution context
+            context = HttpClientContext.create();
+            context.setCredentialsProvider(credsProvider);
+            context.setAuthCache(authCache);
+        } else {
+            context = null;
+        }
+        return context;
+    }
+
+
+    public static <RESPONSE_T, REQUEST_T> RESPONSE_T call( HttpClient taxiiClient
+            , URI endpoint
+            , REQUEST_T request
+            , HttpClientContext context
+            , Class<RESPONSE_T> responseClazz
+    ) throws JAXBException, IOException {
+        //TaxiiXml taxiiXml = xmlFactory.get().createTaxiiXml();
+        //String req = taxiiXml.marshalToString(request, true);
+        // Call the service
+        Object responseObj =  taxiiClient.callTaxiiService(endpoint, request, context);
+        LOG.info("Request made : " + request.getClass().getCanonicalName() + " => " + responseObj.getClass().getCanonicalName() + " (expected " + responseClazz.getCanonicalName() + ")");
+        //String resp = taxiiXml.marshalToString(responseObj, true);
+        try {
+            return responseClazz.cast(responseObj);
+        }
+        catch(ClassCastException cce) {
+            TaxiiXml taxiiXml = xmlFactory.get().createTaxiiXml();
+            String resp = taxiiXml.marshalToString(responseObj, true);
+            String msg = "Didn't return the response we expected: " + responseObj.getClass() + " \n" + resp;
+            LOG.error(msg, cce);
+            throw new RuntimeException(msg, cce);
+        }
+    }
+    private static HttpClient buildClient(URL proxy, String username, String password) throws Exception
+    {
+        HttpClient client = new HttpClient(); // Start with a default TAXII HTTP client.
+
+        // Create an Apache HttpClientBuilder to be customized by the command line arguments.
+        HttpClientBuilder builder = HttpClientBuilder.create().useSystemProperties();
+
+        // Proxy
+        if (proxy != null) {
+            HttpHost proxyHost = new HttpHost(proxy.getHost(), proxy.getPort(), proxy.getProtocol());
+            builder.setProxy(proxyHost);
+        }
+
+        // Basic authentication. User & Password
+        if (username != null ^ password != null) {
+            throw new Exception("'username' and 'password' arguments are required to appear together.");
+        }
+
+
+        // from:  http://stackoverflow.com/questions/19517538/ignoring-ssl-certificate-in-apache-httpclient-4-3
+        SSLContextBuilder ssbldr = new SSLContextBuilder();
+        ssbldr.loadTrustMaterial(null, new TrustSelfSignedStrategy());
+        SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(ssbldr.build(),SSLConnectionSocketFactory.BROWSER_COMPATIBLE_HOSTNAME_VERIFIER);
+
+
+        Registry<ConnectionSocketFactory> registry = RegistryBuilder.<ConnectionSocketFactory>create()
+                .register("http", new PlainConnectionSocketFactory())
+                .register("https", sslsf)
+                .build();
+
+
+        PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(registry);
+        cm.setMaxTotal(20);//max connection
+
+        System.setProperty("jsse.enableSNIExtension", "false"); //""
+        CloseableHttpClient httpClient = builder
+                .setSSLSocketFactory(sslsf)
+                .setConnectionManager(cm)
+                .build();
+
+        client.setHttpclient(httpClient);
+        return client;
+    }
+    public static void main(String... argv) throws Exception {
+        URL endpoint = new URL("http://hailataxii.com/taxii-discovery-service");
+        String username = "guest";
+        String password = "guest";
+        TaxiiConnectionConfig config = new TaxiiConnectionConfig();
+        config = config.withConnectionType(ConnectionType.DISCOVER)
+                       .withEndpoint(endpoint)
+                       .withUsername(username)
+                       .withCollection("guest.Abuse_ch")
+                       .withPassword(password);
+        //TaxiiHandler handler = new TaxiiHandler(config, null);
+        //handler.run();
+        //discoverPollingClient(null, endpoint, username, password, context);
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiLoader.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiLoader.java b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiLoader.java
new file mode 100644
index 0000000..b5385a9
--- /dev/null
+++ b/metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiLoader.java
@@ -0,0 +1,205 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metron.dataloads.nonbulk.taxii;
+
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import org.apache.commons.cli.*;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.util.GenericOptionsParser;
+import org.apache.log4j.PropertyConfigurator;
+import org.apache.metron.dataloads.extractor.Extractor;
+import org.apache.metron.dataloads.extractor.ExtractorHandler;
+import org.apache.metron.dataloads.extractor.stix.StixExtractor;
+import org.apache.metron.common.configuration.EnrichmentConfig;
+import org.apache.metron.common.utils.JSONUtils;
+
+import javax.annotation.Nullable;
+import java.io.File;
+import java.text.*;
+import java.util.Date;
+import java.util.Timer;
+
+public class TaxiiLoader {
+  private static abstract class OptionHandler implements Function<String, Option> {}
+  private enum TaxiiOptions {
+    HELP("h", new OptionHandler() {
+
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        return new Option(s, "help", false, "Generate Help screen");
+      }
+    })
+    ,EXTRACTOR_CONFIG("e", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "extractor_config", true, "JSON Document describing the extractor for this input data source");
+        o.setArgName("JSON_FILE");
+        o.setRequired(true);
+        return o;
+      }
+    })
+    ,CONNECTION_CONFIG("c", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "taxii_connection_config", true, "The JSON config file to configure the connection");
+        o.setArgName("config_file");
+        o.setRequired(true);
+        return o;
+      }
+    })
+    ,TIME_BETWEEN_POLLS("p", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "time_between_polls", true, "The time between polls (in ms)");
+        o.setArgName("MS");
+        o.setRequired(false);
+        return o;
+      }
+    })
+    ,BEGIN_TIME("b", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "begin_time", true, "Start time to poll the Taxii server (all data from that point will be gathered in the first pull).");
+        o.setArgName(DATE_FORMAT.toPattern());
+        o.setRequired(false);
+        return o;
+      }
+    })
+    ,LOG4J_PROPERTIES("l", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "log4j", true, "The log4j properties file to load");
+        o.setArgName("FILE");
+        o.setRequired(false);
+        return o;
+      }
+    })
+    ,ENRICHMENT_CONFIG("n", new OptionHandler() {
+      @Nullable
+      @Override
+      public Option apply(@Nullable String s) {
+        Option o = new Option(s, "enrichment_config", true
+                , "JSON Document describing the enrichment configuration details." +
+                "  This is used to associate an enrichment type with a field type in zookeeper."
+        );
+        o.setArgName("JSON_FILE");
+        o.setRequired(false);
+        return o;
+      }
+    })
+    ;
+    Option option;
+    String shortCode;
+    TaxiiOptions(String shortCode, OptionHandler optionHandler) {
+      this.shortCode = shortCode;
+      this.option = optionHandler.apply(shortCode);
+    }
+
+    public boolean has(CommandLine cli) {
+      return cli.hasOption(shortCode);
+    }
+
+    public String get(CommandLine cli) {
+      return cli.getOptionValue(shortCode);
+    }
+
+    public static CommandLine parse(CommandLineParser parser, String[] args) {
+      try {
+        CommandLine cli = parser.parse(getOptions(), args);
+        if(TaxiiOptions.HELP.has(cli)) {
+          printHelp();
+          System.exit(0);
+        }
+        return cli;
+      } catch (ParseException e) {
+        System.err.println("Unable to parse args: " + Joiner.on(' ').join(args));
+        e.printStackTrace(System.err);
+        printHelp();
+        System.exit(-1);
+        return null;
+      }
+    }
+
+    public static void printHelp() {
+      HelpFormatter formatter = new HelpFormatter();
+      formatter.printHelp( "TaxiiLoader", getOptions());
+    }
+
+    public static Options getOptions() {
+      Options ret = new Options();
+      for(TaxiiOptions o : TaxiiOptions.values()) {
+        ret.addOption(o.option);
+      }
+      return ret;
+    }
+  }
+  public static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+  public static final long ONE_HR_IN_MS = 60*60*1000;
+  public static final long DEFAULT_TIME_BETWEEN_POLLS = ONE_HR_IN_MS;
+
+
+  public static void main(String... argv) throws Exception {
+    Configuration conf = HBaseConfiguration.create();
+    String zkQuorum = conf.get(HConstants.ZOOKEEPER_QUORUM);
+    String[] otherArgs = new GenericOptionsParser(conf, argv).getRemainingArgs();
+
+    CommandLine cli = TaxiiOptions.parse(new PosixParser(), otherArgs);
+    if(TaxiiOptions.LOG4J_PROPERTIES.has(cli)) {
+      PropertyConfigurator.configure(TaxiiOptions.LOG4J_PROPERTIES.get(cli));
+    }
+    ExtractorHandler handler = ExtractorHandler.load(FileUtils.readFileToString(new File(TaxiiOptions.EXTRACTOR_CONFIG.get(cli))));
+    Extractor e = handler.getExtractor();
+    EnrichmentConfig enrichmentConfig = null;
+    if(TaxiiOptions.ENRICHMENT_CONFIG.has(cli)) {
+      enrichmentConfig = JSONUtils.INSTANCE.load( new File(TaxiiOptions.ENRICHMENT_CONFIG.get(cli))
+              , EnrichmentConfig.class
+      );
+      enrichmentConfig.updateSensorConfigs();
+    }
+
+    Timer timer = new Timer();
+    if(e instanceof StixExtractor) {
+      StixExtractor extractor = (StixExtractor)e;
+      TaxiiConnectionConfig connectionConfig = TaxiiConnectionConfig.load(FileUtils.readFileToString(new File(TaxiiOptions.CONNECTION_CONFIG.get(cli))));
+      if(TaxiiOptions.BEGIN_TIME.has(cli)) {
+        Date d = DATE_FORMAT.parse(TaxiiOptions.BEGIN_TIME.get(cli));
+        connectionConfig.withBeginTime(d);
+      }
+      long timeBetween = DEFAULT_TIME_BETWEEN_POLLS;
+      if(TaxiiOptions.TIME_BETWEEN_POLLS.has(cli)) {
+        timeBetween = Long.parseLong(TaxiiOptions.TIME_BETWEEN_POLLS.get(cli));
+      }
+      timer.scheduleAtFixedRate(new TaxiiHandler(connectionConfig, extractor, conf), 0, timeBetween);
+    }
+    else {
+      throw new IllegalStateException("Extractor must be a STIX Extractor");
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerIntegrationTest.java b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerIntegrationTest.java
new file mode 100644
index 0000000..08f95b9
--- /dev/null
+++ b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerIntegrationTest.java
@@ -0,0 +1,153 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.dataloads.bulk;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.metron.TestConstants;
+import org.apache.metron.common.configuration.Configuration;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.indices.IndexMissingException;
+import org.elasticsearch.test.ElasticsearchIntegrationTest;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
+
+@ElasticsearchIntegrationTest.ClusterScope(scope = ElasticsearchIntegrationTest.Scope.SUITE, numDataNodes = 1, numClientNodes = 0)
+public class ElasticsearchDataPrunerIntegrationTest extends ElasticsearchIntegrationTest {
+
+    private static File dataPath = new File("./target/elasticsearch-test");
+    private Date testingDate;
+    private Date yesterday = new Date();
+    private DateFormat dateFormat = new SimpleDateFormat("yyyy.MM.dd.HH");
+    private Configuration configuration;
+
+    @BeforeClass
+    public static void setupClass() throws Exception {
+
+        if (dataPath.isDirectory()) {
+            FileUtils.deleteDirectory(dataPath);
+        }
+
+        if (!dataPath.mkdirs()) {
+            throw new RuntimeException("Couldn't create dataPath at: " + dataPath.getAbsolutePath());
+        }
+
+    }
+
+    @AfterClass
+    public static void teardownClass() throws Exception {
+
+        if (dataPath.isDirectory()) {
+            FileUtils.deleteDirectory(dataPath);
+        }
+
+    }
+
+    @Before
+    public void setUp() throws Exception {
+
+        super.setUp();
+        ensureGreen();
+
+        TimeZone timeZone = TimeZone.getTimeZone("UTC");
+        Calendar calendar = Calendar.getInstance(timeZone);
+        calendar.set(Calendar.HOUR_OF_DAY,0);
+        calendar.set(Calendar.MINUTE,0);
+        calendar.set(Calendar.SECOND,0);
+        testingDate = calendar.getTime();
+        yesterday.setTime(testingDate.getTime() - TimeUnit.DAYS.toMillis(1));
+        dateFormat.setTimeZone(timeZone);
+
+        File resourceFile = new File(TestConstants.SAMPLE_CONFIG_PATH);
+        Path resourcePath = Paths.get(resourceFile.getCanonicalPath());
+
+        configuration = new Configuration(resourcePath);
+    }
+
+    @Test(expected = IndexMissingException.class)
+    public void testWillThrowOnMissingIndex() throws Exception {
+
+        ElasticsearchDataPruner pruner = new ElasticsearchDataPruner(yesterday, 30, configuration,client(), "*");
+        pruner.deleteIndex(admin(), "baz");
+
+    }
+
+    @Test
+    public void testDeletesCorrectIndexes() throws Exception {
+
+        Integer numDays = 5;
+
+        Date createStartDate = new Date();
+
+        createStartDate.setTime(yesterday.getTime() - TimeUnit.DAYS.toMillis(numDays - 1));
+
+        ElasticsearchDataPruner pruner = new ElasticsearchDataPruner(yesterday, 30, configuration,client(), "*");
+        String indexesToDelete = "sensor_index_" + new SimpleDateFormat("yyyy.MM.dd").format(createStartDate) + ".*";
+        Boolean deleted = pruner.deleteIndex(admin(), indexesToDelete);
+
+        assertTrue("Index deletion should be acknowledged", deleted);
+
+    }
+
+    @Test
+    public void testHandlesNoIndicesToDelete() throws Exception {
+
+        ElasticsearchDataPruner pruner = new ElasticsearchDataPruner(yesterday, 1, configuration, client(), "sensor_index_");
+        Long deleteCount = pruner.prune();
+        assertEquals("Should have pruned 0 indices", 0L, deleteCount.longValue());
+
+
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+
+        return ImmutableSettings.settingsBuilder()
+                .put("node.data", true)
+                .put("gateway.type", "none")
+                .put("path.data", dataPath.getPath() + "/data")
+                .put("path.work", dataPath.getPath() + "/work")
+                .put("path.logs", dataPath.getPath() + "/logs")
+                .put("cluster.routing.schedule", "50ms")
+                .put("node.local", true).build();
+
+    }
+
+    public Settings indexSettings() {
+
+        return ImmutableSettings.settingsBuilder()
+                .put("index.store.type", "memory")
+                .put("index.store.fs.memory.enabled", "true")
+                .put("index.number_of_shards", 1)
+                .put("index.number_of_replicas", 0).build();
+
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerRunnerTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerRunnerTest.java b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerRunnerTest.java
new file mode 100644
index 0000000..5f32bee
--- /dev/null
+++ b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerRunnerTest.java
@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.dataloads.bulk;
+
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.FileDescriptor;
+import java.io.FileOutputStream;
+import java.io.PrintStream;
+
+public class ElasticsearchDataPrunerRunnerTest {
+
+    private Options options;
+    private Options help;
+
+    private ByteArrayOutputStream outContent;
+    private ByteArrayOutputStream errContent;
+
+    @Before
+    public void setUp(){
+
+        options = ElasticsearchDataPrunerRunner.buildOptions();
+        help = new Options();
+
+        Option o = new Option("h", "help", false, "This screen");
+        o.setRequired(false);
+        help.addOption(o);
+
+        outContent = new ByteArrayOutputStream();
+        errContent = new ByteArrayOutputStream();
+
+        System.setOut(new PrintStream(outContent));
+        System.setErr(new PrintStream(errContent));
+
+    }
+
+    @Test(expected = RuntimeException.class)
+    public void testThrowsWithoutZookeeperOrConfigLocation() throws Exception {
+
+        String[] args = new String[]{"-n","30","-p","sensor_index","-s","03/30/2016"};
+        ElasticsearchDataPrunerRunner.checkOptions(help,options,args);
+
+    }
+
+    @Test(expected = RuntimeException.class)
+    public void testThrowsWithZookeeperAndConfiguration() throws Exception {
+
+        String[] args = new String[]{"-n","30","-p","sensor_index","-s","03/30/2016"};
+        ElasticsearchDataPrunerRunner.checkOptions(help,options,args);
+
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerTest.java b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerTest.java
new file mode 100644
index 0000000..0cc5d28
--- /dev/null
+++ b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/ElasticsearchDataPrunerTest.java
@@ -0,0 +1,211 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.dataloads.bulk;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.metron.TestConstants;
+import org.apache.metron.common.configuration.Configuration;
+import org.easymock.EasyMock;
+import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateRequestBuilder;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequestBuilder;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
+import org.elasticsearch.client.*;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.common.collect.ImmutableOpenMap;
+import org.elasticsearch.common.hppc.ObjectObjectOpenHashMap;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.indices.IndexMissingException;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Matchers;
+import org.powermock.api.easymock.PowerMock;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.PrintStream;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.powermock.api.easymock.PowerMock.replayAll;
+import static org.powermock.api.easymock.PowerMock.verifyAll;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(DeleteIndexResponse.class)
+public class ElasticsearchDataPrunerTest {
+
+    private Date testDate;
+    private DateFormat dateFormat = new SimpleDateFormat("yyyy.MM.dd.HH");
+    private Configuration configuration;
+
+    private Client indexClient = mock(Client.class);
+    private AdminClient adminClient = mock(AdminClient.class);
+    private IndicesAdminClient indicesAdminClient = mock(FilterClient.IndicesAdmin.class);
+    private DeleteIndexRequestBuilder deleteIndexRequestBuilder = mock(DeleteIndexRequestBuilder.class);
+    private DeleteIndexRequest deleteIndexRequest = mock(DeleteIndexRequest.class);
+    private ActionFuture<DeleteIndexResponse> deleteIndexAction = mock(ActionFuture.class);
+    private DeleteIndexResponse deleteIndexResponse = PowerMock.createMock(DeleteIndexResponse.class);
+
+
+    private ByteArrayOutputStream outContent;
+    private ByteArrayOutputStream errContent;
+
+    @Before
+    public void setUp() throws Exception {
+
+        Calendar calendar = Calendar.getInstance();
+        calendar.set(Calendar.MONTH, Calendar.MARCH);
+        calendar.set(Calendar.YEAR, 2016);
+        calendar.set(Calendar.DATE, 31);
+        calendar.set(Calendar.HOUR_OF_DAY, 0);
+        calendar.set(Calendar.MINUTE, 0);
+        calendar.set(Calendar.SECOND, 0);
+        calendar.set(Calendar.MILLISECOND,0);
+        testDate = calendar.getTime();
+
+        when(indexClient.admin()).thenReturn(adminClient);
+        when(adminClient.indices()).thenReturn(indicesAdminClient);
+        when(indicesAdminClient.prepareDelete(Matchers.<String>anyVararg())).thenReturn(deleteIndexRequestBuilder);
+        when(indicesAdminClient.delete((DeleteIndexRequest) any())).thenReturn(deleteIndexAction);
+        when(deleteIndexRequestBuilder.request()).thenReturn(deleteIndexRequest);
+        when(deleteIndexAction.actionGet()).thenReturn(deleteIndexResponse);
+
+        File resourceFile = new File(TestConstants.SAMPLE_CONFIG_PATH);
+        Path resourcePath = Paths.get(resourceFile.getCanonicalPath());
+
+        configuration = new Configuration(resourcePath);
+
+        outContent = new ByteArrayOutputStream();
+        errContent = new ByteArrayOutputStream();
+
+        System.setOut(new PrintStream(outContent));
+        System.setErr(new PrintStream(errContent));
+
+    }
+
+    @Test(expected = IndexMissingException.class)
+    public void testWillThrowOnMissingIndex() throws Exception {
+
+        when(indicesAdminClient.delete((DeleteIndexRequest) any())).thenThrow(new IndexMissingException(new Index("Test Exception")));
+        ElasticsearchDataPruner pruner = new ElasticsearchDataPruner(testDate, 30, configuration, indexClient,"*");
+        pruner.deleteIndex(adminClient, "baz");
+
+    }
+
+    @Test
+    public void testDeletesCorrectIndexes() throws Exception {
+
+        //Mock Cluster Admin
+        ClusterAdminClient clusterAdminClient = mock(ClusterAdminClient.class);
+        ClusterStateRequestBuilder clusterStateRequestBuilder = mock(ClusterStateRequestBuilder.class);
+        ClusterStateResponse clusterStateResponse = mock(ClusterStateResponse.class);
+        ClusterState clusterState = mock(ClusterState.class);
+        ObjectObjectOpenHashMap<String, IndexMetaData> clusterIndexes = new ObjectObjectOpenHashMap();
+        MetaData clusterMetadata = mock(MetaData.class);
+        when(adminClient.cluster()).thenReturn(clusterAdminClient);
+        when(clusterAdminClient.prepareState()).thenReturn(clusterStateRequestBuilder);
+        when(clusterStateRequestBuilder.get()).thenReturn(clusterStateResponse);
+        when(clusterStateResponse.getState()).thenReturn(clusterState);
+        when(clusterState.getMetaData()).thenReturn(clusterMetadata);
+
+        int numDays = 5;
+
+        Date indexDate = new Date();
+
+        indexDate.setTime(testDate.getTime() - TimeUnit.DAYS.toMillis(numDays));
+
+        for (int i = 0; i < numDays * 24; i++) {
+
+            String indexName = "sensor_index_" + dateFormat.format(indexDate);
+            clusterIndexes.put(indexName, null);
+            indexDate.setTime(indexDate.getTime() + TimeUnit.HOURS.toMillis(1));
+
+        }
+
+        when(clusterMetadata.getIndices()).thenReturn(ImmutableOpenMap.copyOf(clusterIndexes));
+
+
+        EasyMock.expect(deleteIndexResponse.isAcknowledged()).andReturn(true);
+
+        replayAll();
+        ElasticsearchDataPruner pruner = new ElasticsearchDataPruner(testDate, 1, configuration, indexClient, "sensor_index_");
+        pruner.indexClient = indexClient;
+        Long deleteCount = pruner.prune();
+        assertEquals("Should have pruned 24 indices", 24L, deleteCount.longValue());
+        verifyAll();
+
+    }
+
+    @Test
+    public void testFilter() throws Exception {
+
+        ObjectObjectOpenHashMap<String, IndexMetaData> indexNames = new ObjectObjectOpenHashMap();
+        SimpleDateFormat dateChecker = new SimpleDateFormat("yyyyMMdd");
+        int numDays = 5;
+        String[] expectedIndices = new String[24];
+        Date indexDate = new Date();
+
+        indexDate.setTime(testDate.getTime() - TimeUnit.DAYS.toMillis(numDays));
+
+        for (int i = 0, j=0; i < numDays * 24; i++) {
+
+            String indexName = "sensor_index_" + dateFormat.format(indexDate);
+            //Delete 20160330
+            if( dateChecker.format(indexDate).equals("20160330") ){
+                expectedIndices[j++] = indexName;
+            }
+
+            indexNames.put(indexName, null);
+            indexDate.setTime(indexDate.getTime() + TimeUnit.HOURS.toMillis(1));
+
+        }
+
+        ImmutableOpenMap<String, IndexMetaData> testIndices = ImmutableOpenMap.copyOf(indexNames);
+
+        ElasticsearchDataPruner pruner = new ElasticsearchDataPruner(testDate, 1, configuration,  indexClient, "sensor_index_");
+        pruner.indexClient = indexClient;
+
+        Iterable<String> filteredIndices = pruner.getFilteredIndices(testIndices);
+
+        Object[] indexArray = IteratorUtils.toArray(filteredIndices.iterator());
+        Arrays.sort(indexArray);
+        Arrays.sort(expectedIndices);
+
+        assertArrayEquals(expectedIndices,indexArray);
+
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/HDFSDataPrunerTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/HDFSDataPrunerTest.java b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/HDFSDataPrunerTest.java
new file mode 100644
index 0000000..9bc695c
--- /dev/null
+++ b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/bulk/HDFSDataPrunerTest.java
@@ -0,0 +1,178 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.dataloads.bulk;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.*;
+
+
+public class HDFSDataPrunerTest {
+
+
+    private static File dataPath = new File("src/test/resources/HDFSDataPrunerTest");
+
+    private Date todaysDate;
+    private Date yesterday = new Date();
+
+
+    @BeforeClass
+    public static void beforeClass() throws Exception {
+
+        if (dataPath.isDirectory()) {
+            dataPath.delete();
+        }
+
+        if (!dataPath.mkdirs()) {
+            throw new RuntimeException("Couldn't create dataPath at: " + dataPath.getAbsolutePath());
+        }
+
+        dataPath.deleteOnExit();
+
+    }
+
+
+    @Before
+    public void setUp() throws Exception {
+
+        Calendar today = Calendar.getInstance();
+        today.clear(Calendar.HOUR);
+        today.clear(Calendar.MINUTE);
+        today.clear(Calendar.SECOND);
+        todaysDate = today.getTime();
+        yesterday.setTime(todaysDate.getTime() - TimeUnit.DAYS.toMillis(1));
+
+    }
+
+    @Test(expected = StartDateException.class)
+    public void testFailsOnTodaysDate() throws Exception {
+
+        HDFSDataPruner pruner = new HDFSDataPruner(todaysDate, 30, "file:///", dataPath.getAbsolutePath() + "/file-*");
+
+    }
+
+    @Test
+    public void testDeletesCorrectFiles() throws Exception {
+
+        createTestFiles();
+
+        HDFSDataPruner pruner = new HDFSDataPruner(yesterday, 30, "file:///", dataPath.getAbsolutePath() + "/file-*");
+
+        Long prunedCount = pruner.prune();
+        assertTrue("Should have pruned 45 files- pruned: " + prunedCount, 45 == prunedCount);
+
+        //Verify first five files remain
+        File[] filesLeft = dataPath.listFiles();
+        File[] filesList = new File[filesLeft.length];
+        for (int i = 0; i < 5; i++) {
+            filesList[i] = new File(dataPath.getPath() + "//file-" + String.format("%02d", i));
+        }
+        assertArrayEquals("First four files should have been left behind", filesLeft, filesList);
+
+
+    }
+
+    @Test(expected = RuntimeException.class)
+    public void testThrowsIsDirectory() throws Exception {
+
+        FileSystem testFS = mock(FileSystem.class);
+        when(testFS.isDirectory((Path) any())).thenThrow(new IOException("Test Exception"));
+
+        HDFSDataPruner pruner = new HDFSDataPruner(yesterday, 30, "file:///", dataPath.getAbsolutePath() + "/file-*");
+        pruner.fileSystem = testFS;
+        HDFSDataPruner.DateFileFilter filter = pruner.new DateFileFilter(pruner, true);
+
+        filter.accept(new Path("foo"));
+
+    }
+
+    @Test
+    public void testIgnoresDirectoies() throws Exception {
+
+        FileSystem testFS = mock(FileSystem.class);
+        when(testFS.isDirectory((Path) any())).thenReturn(true);
+
+        HDFSDataPruner pruner = new HDFSDataPruner(yesterday, 30, "file:///", dataPath.getAbsolutePath() + "/file-*");
+        pruner.fileSystem = testFS;
+        HDFSDataPruner.DateFileFilter filter = pruner.new DateFileFilter(pruner, false);
+        assertFalse("Should ignore directories",filter.accept(new Path("/tmp")));
+
+    }
+
+    @Test(expected = RuntimeException.class)
+    public void testThrowBadFile() throws Exception {
+
+        FileSystem testFS = mock(FileSystem.class);
+        when(testFS.isDirectory((Path) any())).thenReturn(false);
+        when(testFS.getFileStatus((Path) any())).thenThrow(new IOException("Test Exception"));
+
+        HDFSDataPruner pruner = new HDFSDataPruner(yesterday, 30, "file:///", dataPath.getAbsolutePath() + "/file-*");
+
+        pruner.fileSystem = testFS;
+        HDFSDataPruner.DateFileFilter filter = pruner.new DateFileFilter(pruner, true);
+
+        filter.accept(new Path("foo"));
+
+    }
+
+    private void createTestFiles() throws IOException {
+
+        //create files
+        for (int i = 0; i < 50; i++) {
+            File file = new File(dataPath.getAbsolutePath() + "//file-" + String.format("%02d", i));
+            file.createNewFile();
+            file.deleteOnExit();
+        }
+
+        //Set modification date today - 1 day
+        for (int i = 5; i < 25; i++) {
+            File file = new File(dataPath.getAbsolutePath() + "//file-" + String.format("%02d", i));
+            file.setLastModified(todaysDate.getTime() - TimeUnit.DAYS.toMillis(1));
+            file.deleteOnExit();
+        }
+
+        //Set modification date today - 10 days
+        for (int i = 25; i < 40; i++) {
+            File file = new File(dataPath.getAbsolutePath() + "//file-" + String.format("%02d", i));
+            file.setLastModified(todaysDate.getTime() - TimeUnit.DAYS.toMillis(10));
+            file.deleteOnExit();
+        }
+
+        //Set modification date today - 20 days
+        for (int i = 40; i < 50; i++) {
+            File file = new File(dataPath.getAbsolutePath() + "//file-" + String.format("%02d", i));
+            file.setLastModified(todaysDate.getTime() - TimeUnit.DAYS.toMillis(20));
+            file.deleteOnExit();
+        }
+
+    }
+}
+

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/extractor/ExtractorTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/extractor/ExtractorTest.java b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/extractor/ExtractorTest.java
new file mode 100644
index 0000000..0179193
--- /dev/null
+++ b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/extractor/ExtractorTest.java
@@ -0,0 +1,83 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.dataloads.extractor;
+
+import com.google.common.collect.Iterables;
+import org.apache.metron.enrichment.converter.EnrichmentKey;
+import org.apache.metron.enrichment.converter.EnrichmentValue;
+import org.apache.metron.enrichment.lookup.LookupKV;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+public class ExtractorTest {
+    public static class DummyExtractor implements Extractor
+    {
+
+        @Override
+        public Iterable<LookupKV> extract(String line) throws IOException {
+            EnrichmentKey key = new EnrichmentKey();
+            key.indicator = "dummy";
+            key.type = "type";
+            Map<String, String> value = new HashMap<>();
+            value.put("indicator", "dummy");
+            return Arrays.asList(new LookupKV(key, new EnrichmentValue(value)));
+        }
+
+        @Override
+        public void initialize(Map<String, Object> config) {
+
+        }
+    }
+    @Test
+    public void testDummyExtractor() throws IllegalAccessException, InstantiationException, ClassNotFoundException, IOException {
+        Extractor extractor = Extractors.create(DummyExtractor.class.getName());
+        LookupKV results = Iterables.getFirst(extractor.extract(null), null);
+        EnrichmentKey key = (EnrichmentKey) results.getKey();
+        EnrichmentValue value = (EnrichmentValue) results.getValue();
+        Assert.assertEquals("dummy", key.indicator);
+        Assert.assertEquals("type", key.type);
+        Assert.assertEquals("dummy", value.getMetadata().get("indicator"));
+    }
+
+    @Test
+    public void testExtractionLoading() throws Exception {
+        /**
+         config:
+         {
+            "config" : {}
+            ,"extractor" : "org.apache.metron.dataloads.extractor.ExtractorTest$DummyExtractor"
+         }
+         */
+        String config = "{\n" +
+                "            \"config\" : {}\n" +
+                "            ,\"extractor\" : \"org.apache.metron.dataloads.extractor.ExtractorTest$DummyExtractor\"\n" +
+                "         }";
+        ExtractorHandler handler = ExtractorHandler.load(config);
+        LookupKV results = Iterables.getFirst(handler.getExtractor().extract(null), null);
+        EnrichmentKey key = (EnrichmentKey) results.getKey();
+        EnrichmentValue value = (EnrichmentValue) results.getValue();
+        Assert.assertEquals("dummy", key.indicator);
+        Assert.assertEquals("type", key.type);
+        Assert.assertEquals("dummy", value.getMetadata().get("indicator"));
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/0117987e/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/extractor/csv/CSVExtractorTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/extractor/csv/CSVExtractorTest.java b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/extractor/csv/CSVExtractorTest.java
new file mode 100644
index 0000000..6cd82c7
--- /dev/null
+++ b/metron-platform/metron-data-management/src/test/java/org/apache/metron/dataloads/extractor/csv/CSVExtractorTest.java
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.dataloads.extractor.csv;
+
+import com.google.common.collect.Iterables;
+import org.adrianwalker.multilinestring.Multiline;
+import org.apache.metron.dataloads.extractor.ExtractorHandler;
+import org.apache.metron.enrichment.converter.EnrichmentKey;
+import org.apache.metron.enrichment.converter.EnrichmentValue;
+import org.apache.metron.enrichment.lookup.LookupKV;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class CSVExtractorTest {
+
+  /**
+   {
+     "config" : {
+        "columns" : {
+            "host" : 0
+           ,"meta" : 2
+                    }
+       ,"indicator_column" : "host"
+       ,"type" : "threat"
+       ,"separator" : ","
+               }
+     ,"extractor" : "CSV"
+   }
+   */
+  @Multiline
+  static String testCSVConfig;
+
+  @Test
+  public void testCSVExtractor() throws Exception {
+
+    ExtractorHandler handler = ExtractorHandler.load(testCSVConfig);
+    validate(handler);
+  }
+
+  public void validate(ExtractorHandler handler) throws IOException {
+    {
+      LookupKV results = Iterables.getFirst(handler.getExtractor().extract("google.com,1.0,foo"), null);
+      EnrichmentKey key = (EnrichmentKey) results.getKey();
+      EnrichmentValue value = (EnrichmentValue) results.getValue();
+      Assert.assertEquals("google.com", key.indicator);
+      Assert.assertEquals("threat", key.type);
+      Assert.assertEquals("google.com", value.getMetadata().get("host"));
+      Assert.assertEquals("foo", value.getMetadata().get("meta"));
+      Assert.assertEquals(2, value.getMetadata().size());
+    }
+    {
+      Iterable<LookupKV> results = handler.getExtractor().extract("#google.com,1.0,foo");
+      Assert.assertEquals(0, Iterables.size(results));
+    }
+  }
+}