metron-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmerri...@apache.org
Subject [2/3] incubator-metron git commit: METRON-694 Index Errors from Topologies (merrimanr) closes apache/incubator-metron#453
Date Mon, 06 Mar 2017 21:25:49 GMT
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java
index c5c1294..1f33060 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/BulkMessageWriterBoltTest.java
@@ -18,6 +18,8 @@
 package org.apache.metron.enrichment.bolt;
 
 import org.apache.log4j.Level;
+import org.apache.metron.common.message.MessageGetStrategy;
+import org.apache.metron.common.message.MessageGetters;
 import org.apache.metron.common.writer.BulkWriterResponse;
 import org.apache.metron.test.utils.UnitTestHelper;
 import org.apache.metron.writer.BulkWriterComponent;
@@ -112,9 +114,13 @@ public class BulkMessageWriterBoltTest extends BaseEnrichmentBoltTest {
   @Mock
   private BulkMessageWriter<JSONObject> bulkMessageWriter;
 
+  @Mock
+  private MessageGetStrategy messageGetStrategy;
+
   @Test
   public void test() throws Exception {
-    BulkMessageWriterBolt bulkMessageWriterBolt = new BulkMessageWriterBolt("zookeeperUrl").withBulkMessageWriter(bulkMessageWriter);
+    BulkMessageWriterBolt bulkMessageWriterBolt = new BulkMessageWriterBolt("zookeeperUrl")
+            .withBulkMessageWriter(bulkMessageWriter).withMessageGetter(MessageGetters.JSON_FROM_FIELD.name()).withMessageGetterField("message");
     bulkMessageWriterBolt.setCuratorFramework(client);
     bulkMessageWriterBolt.setTreeCache(cache);
     bulkMessageWriterBolt.getConfigurations().updateSensorIndexingConfig(sensorType, new FileInputStream(sampleSensorIndexingConfigPath));

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java
index b0076a4..90322fe 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBoltTest.java
@@ -17,18 +17,22 @@
  */
 package org.apache.metron.enrichment.bolt;
 
-import org.apache.log4j.Level;
-import org.apache.metron.enrichment.adapters.geo.GeoLiteDatabase;
-import org.apache.metron.test.utils.UnitTestHelper;
-import org.apache.storm.tuple.Values;
+import com.google.common.cache.CacheLoader;
 import com.google.common.collect.ImmutableMap;
 import org.adrianwalker.multilinestring.Multiline;
+import org.apache.log4j.Level;
 import org.apache.metron.TestConstants;
-import org.apache.metron.test.bolt.BaseEnrichmentBoltTest;
-import org.apache.metron.enrichment.configuration.Enrichment;
+import org.apache.metron.common.Constants;
+import org.apache.metron.common.configuration.ConfigurationsUtils;
 import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
+import org.apache.metron.common.error.MetronError;
+import org.apache.metron.enrichment.adapters.geo.GeoLiteDatabase;
+import org.apache.metron.enrichment.configuration.Enrichment;
 import org.apache.metron.enrichment.interfaces.EnrichmentAdapter;
-import org.apache.metron.common.configuration.ConfigurationsUtils;
+import org.apache.metron.test.bolt.BaseEnrichmentBoltTest;
+import org.apache.metron.test.error.MetronErrorJSONMatcher;
+import org.apache.metron.test.utils.UnitTestHelper;
+import org.apache.storm.tuple.Values;
 import org.hamcrest.Description;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
@@ -43,12 +47,16 @@ import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.HashSet;
 
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.argThat;
 import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class GenericEnrichmentBoltTest extends BaseEnrichmentBoltTest {
 
@@ -194,7 +202,10 @@ public class GenericEnrichmentBoltTest extends BaseEnrichmentBoltTest {
     UnitTestHelper.setLog4jLevel(GenericEnrichmentBolt.class, Level.FATAL);
     genericEnrichmentBolt.execute(tuple);
     UnitTestHelper.setLog4jLevel(GenericEnrichmentBolt.class, Level.ERROR);
-    verify(outputCollector, times(1)).emit(eq("error"), any(Values.class));
+    MetronError error = new MetronError()
+            .withErrorType(Constants.ErrorType.ENRICHMENT_ERROR)
+            .withThrowable(new Exception("Could not parse binary stream to JSON"));
+    verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), argThat(new MetronErrorJSONMatcher(error.getJSONObject())));
     when(tuple.getStringByField("key")).thenReturn(key);
     when(tuple.getValueByField("message")).thenReturn(originalMessage);
     when(enrichmentAdapter.enrich(any())).thenReturn(new JSONObject());
@@ -217,6 +228,19 @@ public class GenericEnrichmentBoltTest extends BaseEnrichmentBoltTest {
     verify(enrichmentAdapter, times(1)).logAccess(cacheKey2);
     verify(outputCollector, times(1)).emit(eq(enrichmentType), argThat(new EnrichedMessageMatcher(key, enrichedMessage)));
 
-
+    reset(outputCollector);
+    genericEnrichmentBolt.cache.invalidateAll();
+    when(enrichmentAdapter.enrich(cacheKey1)).thenReturn(null);
+    genericEnrichmentBolt.execute(tuple);
+    error = new MetronError()
+            .withErrorType(Constants.ErrorType.ENRICHMENT_ERROR)
+            .withErrorFields(new HashSet<String>() {{ add("field1"); }})
+            .addRawMessage(new JSONObject() {{
+              put("field1", "value1");
+              put("field2", "value2");
+              put("source.type", "test");
+            }})
+            .withThrowable(new CacheLoader.InvalidCacheLoadException("CacheLoader returned null for key CacheKey{field='field1', value='value1'}."));
+    verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), argThat(new MetronErrorJSONMatcher(error.getJSONObject())));
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java
index 5b06d33..9f12fcd 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java
@@ -17,10 +17,14 @@
  */
 package org.apache.metron.enrichment.bolt;
 
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.tuple.Values;
+import com.google.common.cache.LoadingCache;
 import org.adrianwalker.multilinestring.Multiline;
+import org.apache.metron.common.Constants;
+import org.apache.metron.common.error.MetronError;
 import org.apache.metron.test.bolt.BaseEnrichmentBoltTest;
+import org.apache.metron.test.error.MetronErrorJSONMatcher;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.tuple.Values;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
 import org.json.simple.parser.ParseException;
@@ -31,11 +35,13 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.argThat;
 import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -89,7 +95,7 @@ public class JoinBoltTest extends BaseEnrichmentBoltTest {
   }
 
   @Test
-  public void test() {
+  public void test() throws Exception {
     StandAloneJoinBolt joinBolt = new StandAloneJoinBolt("zookeeperUrl");
     joinBolt.setCuratorFramework(client);
     joinBolt.setTreeCache(cache);
@@ -126,5 +132,16 @@ public class JoinBoltTest extends BaseEnrichmentBoltTest {
     joinBolt.execute(tuple);
     verify(outputCollector, times(1)).emit(eq("message"), any(tuple.getClass()), eq(new Values(key, joinedMessage)));
     verify(outputCollector, times(1)).ack(tuple);
+
+    joinBolt.cache = mock(LoadingCache.class);
+    when(joinBolt.cache.get(key)).thenThrow(new ExecutionException(new Exception("join exception")));
+    joinBolt.execute(tuple);
+
+    MetronError error = new MetronError()
+            .withErrorType(Constants.ErrorType.ENRICHMENT_ERROR)
+            .withMessage("Joining problem: {}")
+            .withThrowable(new ExecutionException(new Exception("join exception")))
+            .addRawMessage(new JSONObject());
+    verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), argThat(new MetronErrorJSONMatcher(error.getJSONObject())));
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java
index 7306c64..e012c55 100644
--- a/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java
+++ b/metron-platform/metron-enrichment/src/test/java/org/apache/metron/enrichment/integration/EnrichmentIntegrationTest.java
@@ -76,6 +76,7 @@ import java.util.stream.Stream;
 import static org.apache.metron.enrichment.bolt.ThreatIntelJoinBolt.*;
 
 public class EnrichmentIntegrationTest extends BaseIntegrationTest {
+  private static final String ERROR_TOPIC = "enrichment_error";
   private static final String SRC_IP = "ip_src_addr";
   private static final String DST_IP = "ip_dst_addr";
   private static final String MALICIOUS_IP_TYPE = "malicious_ip";
@@ -139,13 +140,13 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
       setProperty("enrichment.simple.hbase.table", enrichmentsTableName);
       setProperty("enrichment.simple.hbase.cf", cf);
       setProperty("enrichment.output.topic", Constants.INDEXING_TOPIC);
-      setProperty("enrichment.error.topic", Constants.ENRICHMENT_ERROR_TOPIC);
+      setProperty("enrichment.error.topic", ERROR_TOPIC);
     }};
     final ZKServerComponent zkServerComponent = getZKServerComponent(topologyProperties);
     final KafkaComponent kafkaComponent = getKafkaComponent(topologyProperties, new ArrayList<KafkaComponent.Topic>() {{
       add(new KafkaComponent.Topic(Constants.ENRICHMENT_TOPIC, 1));
       add(new KafkaComponent.Topic(Constants.INDEXING_TOPIC, 1));
-      add(new KafkaComponent.Topic(Constants.ENRICHMENT_ERROR_TOPIC, 1));
+      add(new KafkaComponent.Topic(ERROR_TOPIC, 1));
     }});
     String globalConfigStr = null;
     {
@@ -201,15 +202,14 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
       fluxComponent.submitTopology();
 
       kafkaComponent.writeMessages(Constants.ENRICHMENT_TOPIC, inputMessages);
-      ProcessorResult<List<Map<String, Object>>> result = runner.process(getProcessor());
-      // We expect failures, so we don't care if result returned failure or not
-      List<Map<String, Object>> docs = result.getResult();
+      ProcessorResult<Map<String, List<Map<String, Object>>>> result = runner.process(getProcessor());
+      Map<String,List<Map<String, Object>>> outputMessages = result.getResult();
+      List<Map<String, Object>> docs = outputMessages.get(Constants.INDEXING_TOPIC);
       Assert.assertEquals(inputMessages.size(), docs.size());
       validateAll(docs);
-
-      List<byte[]> errors = result.getProcessErrors();
+      List<Map<String, Object>> errors = outputMessages.get(ERROR_TOPIC);
       Assert.assertEquals(inputMessages.size(), errors.size());
-      validateErrors(result.getProcessErrors());
+      validateErrors(errors);
     } finally {
       runner.stop();
     }
@@ -234,10 +234,12 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
     }
   }
 
-  protected void validateErrors(List<byte[]> errors) {
-    for(byte[] error : errors) {
-      // Don't reconstruct the entire message, just ensure it contains the known error message inside.
-      Assert.assertTrue(new String(error).contains(ErrorEnrichmentBolt.TEST_ERROR_MESSAGE));
+  protected void validateErrors(List<Map<String, Object>> errors) {
+    for(Map<String, Object> error : errors) {
+      Assert.assertEquals("Test throwing error from ErrorEnrichmentBolt", error.get(Constants.ErrorFields.MESSAGE.getName()));
+      Assert.assertEquals("java.lang.IllegalStateException: Test throwing error from ErrorEnrichmentBolt", error.get(Constants.ErrorFields.EXCEPTION.getName()));
+      Assert.assertEquals(Constants.ErrorType.ENRICHMENT_ERROR.getType(), error.get(Constants.ErrorFields.ERROR_TYPE.getName()));
+      Assert.assertEquals("{\"rawMessage\":\"Error Test Raw Message String\"}", error.get(Constants.ErrorFields.RAW_MESSAGE.getName()));
     }
   }
 
@@ -504,39 +506,47 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
     return ret;
   }
 
+  private static List<Map<String, Object>> loadMessages(List<byte[]> outputMessages) {
+    List<Map<String, Object>> tmp = new ArrayList<>();
+    Iterables.addAll(tmp
+            , Iterables.transform(outputMessages
+                    , message -> {
+                      try {
+                        return new HashMap<>(JSONUtils.INSTANCE.load(new String(message)
+                                , new TypeReference<Map<String, Object>>() {}
+                        )
+                        );
+                      } catch (Exception ex) {
+                        throw new IllegalStateException(ex);
+                      }
+                    }
+            )
+    );
+    return tmp;
+  }
   @SuppressWarnings("unchecked")
-  private Processor<List<Map<String, Object>>> getProcessor() {
+  private KafkaProcessor<Map<String,List<Map<String, Object>>>> getProcessor(){
 
-    KafkaProcessor<List<Map<String, Object>>> kafkaProcessor = new KafkaProcessor<>().withKafkaComponentName("kafka")
+    return new KafkaProcessor<>()
+            .withKafkaComponentName("kafka")
             .withReadTopic(Constants.INDEXING_TOPIC)
-            .withErrorTopic(Constants.ENRICHMENT_ERROR_TOPIC)
-            .withInvalidTopic(Constants.INVALID_STREAM)
+            .withErrorTopic(ERROR_TOPIC)
             .withValidateReadMessages(new Function<KafkaMessageSet, Boolean>() {
               @Nullable
               @Override
               public Boolean apply(@Nullable KafkaMessageSet messageSet) {
-                // this test is written to return 10 errors and 10 messages
-                // we can just check when the messages match here
-                // if they do then we are good
-                return messageSet.getMessages().size() == inputMessages.size();
+                return (messageSet.getMessages().size() == inputMessages.size()) && (messageSet.getErrors().size() == inputMessages.size());
               }
             })
-            .withProvideResult(new Function<KafkaMessageSet , List<Map<String, Object>>>() {
+            .withProvideResult(new Function<KafkaMessageSet,Map<String,List<Map<String, Object>>>>(){
               @Nullable
               @Override
-              public List<Map<String, Object>> apply(@Nullable KafkaMessageSet messageSet) {
-                List<Map<String,Object>> docs = new ArrayList<>();
-                for (byte[] message : messageSet.getMessages()) {
-                  try {
-                    docs.add(JSONUtils.INSTANCE.load(new String(message), new TypeReference<Map<String, Object>>() {
-                    }));
-                  } catch (IOException e) {
-                    throw new IllegalStateException(e.getMessage(), e);
-                  }
-                }
-                return docs;
+              public Map<String,List<Map<String, Object>>> apply(@Nullable KafkaMessageSet messageSet) {
+                return new HashMap<String, List<Map<String, Object>>>() {{
+                  put(Constants.INDEXING_TOPIC, loadMessages(messageSet.getMessages()));
+                  put(ERROR_TOPIC, loadMessages(messageSet.getErrors()));
+                }};
               }
             });
-    return kafkaProcessor;
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-indexing/README.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/README.md b/metron-platform/metron-indexing/README.md
index 5296ea0..f243eef 100644
--- a/metron-platform/metron-indexing/README.md
+++ b/metron-platform/metron-indexing/README.md
@@ -24,7 +24,7 @@ and sent to
 * An indexing bolt configured to write to either elasticsearch or Solr
 * An indexing bolt configured to write to HDFS under `/apps/metron/enrichment/indexed`
 
-Errors during indexing are sent to a kafka queue called `index_errors`
+By default, errors during indexing are sent back into the `indexing` kafka queue so that they can be indexed and archived.
 
 ##Sensor Indexing Configuration
 The sensor specific configuration is intended to configure the

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-indexing/src/main/config/zookeeper/indexing/error.json
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/config/zookeeper/indexing/error.json b/metron-platform/metron-indexing/src/main/config/zookeeper/indexing/error.json
new file mode 100644
index 0000000..4c9d786
--- /dev/null
+++ b/metron-platform/metron-indexing/src/main/config/zookeeper/indexing/error.json
@@ -0,0 +1,17 @@
+{
+  "hdfs" : {
+    "index": "error",
+    "batchSize": 5,
+    "enabled" : true
+  },
+  "elasticsearch" : {
+    "index": "error",
+    "batchSize": 5,
+    "enabled" : true
+  },
+  "solr" : {
+    "index": "error",
+    "batchSize": 5,
+    "enabled" : true
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-indexing/src/main/flux/indexing/remote.yaml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/main/flux/indexing/remote.yaml b/metron-platform/metron-indexing/src/main/flux/indexing/remote.yaml
index 8bf8f48..3e329f4 100644
--- a/metron-platform/metron-indexing/src/main/flux/indexing/remote.yaml
+++ b/metron-platform/metron-indexing/src/main/flux/indexing/remote.yaml
@@ -101,7 +101,8 @@ bolts:
                     - ref: "indexWriter"
             -   name: "withMessageGetter"
                 args:
-                    - "RAW"
+                    - "DEFAULT_JSON_FROM_POSITION"
+
     -   id: "hdfsIndexingBolt"
         className: "org.apache.metron.writer.bolt.BulkMessageWriterBolt"
         constructorArgs:
@@ -112,7 +113,7 @@ bolts:
                     - ref: "hdfsWriter"
             -   name: "withMessageGetter"
                 args:
-                    - "RAW"
+                    - "DEFAULT_JSON_FROM_POSITION"
 
     -   id: "indexingErrorBolt"
         className: "org.apache.metron.writer.bolt.BulkMessageWriterBolt"

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/IndexingIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/IndexingIntegrationTest.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/IndexingIntegrationTest.java
index ae04e43..60cd1d1 100644
--- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/IndexingIntegrationTest.java
+++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/integration/IndexingIntegrationTest.java
@@ -44,7 +44,6 @@ import org.junit.Assert;
 import org.junit.Test;
 
 import javax.annotation.Nullable;
-import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.IOException;
 import java.util.*;
@@ -53,6 +52,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import static org.apache.metron.common.configuration.ConfigurationsUtils.getClient;
 
 public abstract class IndexingIntegrationTest extends BaseIntegrationTest {
+  protected static final String ERROR_TOPIC = "indexing_error";
   protected String hdfsDir = "target/indexingIntegrationTest/hdfs";
   protected String sampleParsedPath = TestConstants.SAMPLE_DATA_PARSED_PATH + "TestExampleParsed";
   protected String fluxPath = "../metron-indexing/src/main/flux/indexing/remote.yaml";
@@ -125,7 +125,7 @@ public abstract class IndexingIntegrationTest extends BaseIntegrationTest {
       setProperty("indexing.workers", "1");
       setProperty("indexing.executors", "0");
       setProperty("index.input.topic", Constants.INDEXING_TOPIC);
-      setProperty("index.error.topic", Constants.INDEXING_ERROR_TOPIC);
+      setProperty("index.error.topic", ERROR_TOPIC);
       setProperty("index.date.format", dateFormat);
       //HDFS settings
 
@@ -138,7 +138,7 @@ public abstract class IndexingIntegrationTest extends BaseIntegrationTest {
     final ZKServerComponent zkServerComponent = getZKServerComponent(topologyProperties);
     final KafkaComponent kafkaComponent = getKafkaComponent(topologyProperties, new ArrayList<KafkaComponent.Topic>() {{
       add(new KafkaComponent.Topic(Constants.INDEXING_TOPIC, 1));
-      add(new KafkaComponent.Topic(Constants.INDEXING_ERROR_TOPIC, 1));
+      add(new KafkaComponent.Topic(ERROR_TOPIC, 1));
     }});
     List<Map<String, Object>> inputDocs = new ArrayList<>();
     for(byte[] b : inputMessages) {

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ProcessorResult.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ProcessorResult.java b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ProcessorResult.java
index 3eb4e8f..c309686 100644
--- a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ProcessorResult.java
+++ b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ProcessorResult.java
@@ -24,7 +24,6 @@ public class ProcessorResult<T> {
     public static class Builder<T>{
         T result;
         List<byte[]> processErrors;
-        List<byte[]> processInvalids;
 
         public Builder(){}
 
@@ -38,25 +37,18 @@ public class ProcessorResult<T> {
             return this;
         }
 
-        public Builder withProcessInvalids(List<byte[]> processInvalids){
-            this.processInvalids = processInvalids;
-            return this;
-        }
-
         public ProcessorResult<T> build(){
-            return new ProcessorResult<T>(result,processErrors,processInvalids);
+            return new ProcessorResult<T>(result,processErrors);
         }
 
     }
 
     T result;
     List<byte[]> processErrors;
-    List<byte[]> processInvalids;
     @SuppressWarnings("unchecked")
-    public ProcessorResult(T result,List<byte[]> processErrors, List<byte[]> processInvalids){
+    public ProcessorResult(T result,List<byte[]> processErrors){
         this.result = result;
         this.processErrors = processErrors == null ? new ArrayList() : processErrors;
-        this.processInvalids = processInvalids == null ? new ArrayList() : processInvalids;
     }
 
     public T getResult(){
@@ -67,12 +59,8 @@ public class ProcessorResult<T> {
         return processErrors;
     }
 
-    public List<byte[]> getProcessInvalids(){
-        return processInvalids;
-    }
-
     public boolean failed(){
-        return processErrors.size() > 0 || processInvalids.size() > 0;
+        return processErrors.size() > 0;
     }
 
     public void getBadResults(StringBuffer buffer){
@@ -84,10 +72,5 @@ public class ProcessorResult<T> {
             buffer.append(new String(outputMessage));
         }
         buffer.append("\n");
-        buffer.append(String.format("%d Invalid Messages", processInvalids.size()));
-        for (byte[] outputMessage : processInvalids) {
-            buffer.append(new String(outputMessage));
-        }
-        buffer.append("\n");
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaMessageSet.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaMessageSet.java b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaMessageSet.java
index 4227933..683fe6a 100644
--- a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaMessageSet.java
+++ b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaMessageSet.java
@@ -23,12 +23,10 @@ import java.util.List;
 public class KafkaMessageSet{
     public List<byte[]> messages;
     public List<byte[]> errors;
-    public List<byte[]> invalids;
 
-    public KafkaMessageSet(List<byte[]> messages, List<byte[]> errors, List<byte[]> invalids) {
+    public KafkaMessageSet(List<byte[]> messages, List<byte[]> errors) {
         this.messages = messages;
         this.errors = errors;
-        this.invalids = invalids;
     }
 
 
@@ -38,7 +36,4 @@ public class KafkaMessageSet{
     public List<byte[]> getErrors() {
         return errors;
     }
-    public List<byte[]> getInvalids() {
-        return invalids;
-    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaProcessor.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaProcessor.java b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaProcessor.java
index 6fdbbf4..63f073d 100644
--- a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaProcessor.java
+++ b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/processors/KafkaProcessor.java
@@ -30,10 +30,8 @@ public class KafkaProcessor<T> implements Processor<T> {
     private String kafkaComponentName;
     private String readTopic;
     private String errorTopic;
-    private String invalidTopic;
     private List<byte[]> messages = new LinkedList<>();
     private List<byte[]> errors = new LinkedList<>();
-    private List<byte[]> invalids = new LinkedList<>();
 
     public KafkaProcessor(){}
     public KafkaProcessor withKafkaComponentName(String name){
@@ -48,10 +46,6 @@ public class KafkaProcessor<T> implements Processor<T> {
         this.errorTopic = topicName;
         return this;
     }
-    public KafkaProcessor withInvalidTopic(String topicName){
-        this.invalidTopic = topicName;
-        return this;
-    }
     public KafkaProcessor withValidateReadMessages(Function<KafkaMessageSet, Boolean> validate){
         this.validateReadMessages = validate;
         return this;
@@ -68,25 +62,19 @@ public class KafkaProcessor<T> implements Processor<T> {
         KafkaComponent kafkaComponent = runner.getComponent(kafkaComponentName, KafkaComponent.class);
         LinkedList<byte[]> outputMessages = new LinkedList<>(kafkaComponent.readMessages(readTopic));
         LinkedList<byte[]> outputErrors = null;
-        LinkedList<byte[]> outputInvalids = null;
 
         if (errorTopic != null) {
             outputErrors = new LinkedList<>(kafkaComponent.readMessages(errorTopic));
         }
-        if (invalidTopic != null) {
-            outputInvalids = new LinkedList<>(kafkaComponent.readMessages(invalidTopic));
-        }
-        Boolean validated = validateReadMessages.apply(new KafkaMessageSet(outputMessages,outputErrors,outputInvalids));
+        Boolean validated = validateReadMessages.apply(new KafkaMessageSet(outputMessages,outputErrors));
         if(validated == null){
             validated = false;
         }
         if(validated){
             messages.addAll(outputMessages);
             errors.addAll(outputErrors);
-            invalids.addAll(outputInvalids);
             outputMessages.clear();
             outputErrors.clear();
-            outputInvalids.clear();
             return ReadinessState.READY;
         }
         return ReadinessState.NOT_READY;
@@ -94,7 +82,7 @@ public class KafkaProcessor<T> implements Processor<T> {
     @SuppressWarnings("unchecked")
     public ProcessorResult<T> getResult(){
         ProcessorResult.Builder<T> builder = new ProcessorResult.Builder();
-        return builder.withResult(provideResult.apply(new KafkaMessageSet(messages,errors,invalids))).withProcessErrors(errors).withProcessInvalids(invalids).build();
+        return builder.withResult(provideResult.apply(new KafkaMessageSet(messages,errors))).withProcessErrors(errors).build();
     }
 }
 

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/README.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/README.md b/metron-platform/metron-parsers/README.md
index 3c4310d..2cf9bbf 100644
--- a/metron-platform/metron-parsers/README.md
+++ b/metron-platform/metron-parsers/README.md
@@ -30,7 +30,7 @@ Data flows through the parser bolt via kafka and into the `enrichments`
 topology in kafka.  Errors are collected with the context of the error
 (e.g. stacktrace) and original message causing the error and sent to an
 `error` queue.  Invalid messages as determined by global validation
-functions are sent to an `invalid` queue. 
+functions are also treated as errors and sent to an `error` queue. 
  
 ##Message Format
 
@@ -277,9 +277,6 @@ usage: start_parser_topology.sh
  -ewp,--error_writer_p <PARALLELISM_HINT>       Error Writer Parallelism
                                                 Hint
  -h,--help                                      This screen
- -iwnt,--invalid_writer_num_tasks <NUM_TASKS>   Invalid Writer Num Tasks
- -iwp,--invalid_writer_p <PARALLELISM_HINT>     Invalid Message Writer
-                                                Parallelism Hint
  -k,--kafka <BROKER_URL>                        Kafka Broker URL
  -mt,--message_timeout <TIMEOUT_IN_SECS>        Message Timeout in Seconds
  -mtp,--max_task_parallelism <MAX_TASK>         Max task parallelism
@@ -365,9 +362,6 @@ be customized by modifying the arguments sent to this utility.
 * The Error Message Writer Bolt
   * `--error_writer_num_tasks` : The number of tasks for the error writer bolt
   * `--error_writer_p` : The parallelism hint for the error writer bolt
-* The Invalid Message Writer Bolt
-  * `--invalid_writer_num_tasks` : The number of tasks for the error writer bolt
-  * `--invalid_writer_p` : The parallelism hint for the error writer bolt
  
 Finally, if workers and executors are new to you, the following might be of use to you:
 * [Understanding the Parallelism of a Storm Topology](http://www.michael-noll.com/blog/2012/10/16/understanding-the-parallelism-of-a-storm-topology/)

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java
index 20c15ce..0b1ea3d 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java
@@ -18,29 +18,38 @@
 package org.apache.metron.parsers.bolt;
 
 import org.apache.commons.lang3.StringUtils;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
-import org.apache.storm.tuple.Values;
 import org.apache.metron.common.Constants;
 import org.apache.metron.common.bolt.ConfiguredParserBolt;
+import org.apache.metron.common.configuration.FieldTransformer;
 import org.apache.metron.common.configuration.FieldValidator;
 import org.apache.metron.common.configuration.SensorParserConfig;
 import org.apache.metron.common.dsl.Context;
 import org.apache.metron.common.dsl.StellarFunctions;
-import org.apache.metron.parsers.filters.Filters;
-import org.apache.metron.common.configuration.FieldTransformer;
+import org.apache.metron.common.error.MetronError;
+import org.apache.metron.common.message.MessageGetStrategy;
+import org.apache.metron.common.message.MessageGetters;
 import org.apache.metron.common.utils.ErrorUtils;
+import org.apache.metron.enrichment.adapters.geo.GeoLiteDatabase;
+import org.apache.metron.parsers.filters.Filters;
 import org.apache.metron.parsers.interfaces.MessageFilter;
 import org.apache.metron.parsers.interfaces.MessageParser;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
 import org.json.simple.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 public class ParserBolt extends ConfiguredParserBolt implements Serializable {
 
@@ -51,6 +60,7 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
   private MessageFilter<JSONObject> filter;
   private WriterHandler writer;
   private org.apache.metron.common.dsl.Context stellarContext;
+  private transient MessageGetStrategy messageGetStrategy;
   public ParserBolt( String zookeeperUrl
                    , String sensorType
                    , MessageParser<JSONObject> parser
@@ -72,6 +82,7 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
   @Override
   public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
     super.prepare(stormConf, context, collector);
+    messageGetStrategy = MessageGetters.DEFAULT_BYTES_FROM_POSITION.get();
     this.collector = collector;
     initializeStellar();
     if(getSensorParserConfig() != null && filter == null) {
@@ -109,7 +120,7 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
   @SuppressWarnings("unchecked")
   @Override
   public void execute(Tuple tuple) {
-    byte[] originalMessage = tuple.getBinary(0);
+    byte[] originalMessage = (byte[]) messageGetStrategy.get(tuple);
     SensorParserConfig sensorParserConfig = getSensorParserConfig();
     try {
       //we want to ack the tuple in the situation where we have are not doing a bulk write
@@ -128,12 +139,22 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
           }
           if (parser.validate(message) && (filter == null || filter.emitTuple(message, stellarContext))) {
             numWritten++;
-            if(!isGloballyValid(message, fieldValidations)) {
-              message.put(Constants.SENSOR_TYPE, getSensorType()+ ".invalid");
-              collector.emit(Constants.INVALID_STREAM, new Values(message));
+            List<FieldValidator> failedValidators = getFailedValidators(message, fieldValidations);
+            if(failedValidators.size() > 0) {
+              MetronError error = new MetronError()
+                      .withErrorType(Constants.ErrorType.PARSER_INVALID)
+                      .withSensorType(getSensorType())
+                      .addRawMessage(message);
+              Set<String> errorFields = failedValidators.stream()
+                      .flatMap(fieldValidator -> fieldValidator.getInput().stream())
+                      .collect(Collectors.toSet());
+              if (!errorFields.isEmpty()) {
+                error.withErrorFields(errorFields);
+              }
+              ErrorUtils.handleError(collector, error);
             }
             else {
-              writer.write(getSensorType(), tuple, message, getConfigurations());
+              writer.write(getSensorType(), tuple, message, getConfigurations(), messageGetStrategy);
             }
           }
         }
@@ -145,28 +166,28 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable {
         collector.ack(tuple);
       }
     } catch (Throwable ex) {
-      ErrorUtils.handleError( collector
-                            , ex
-                            , Constants.ERROR_STREAM
-                            , Optional.of(getSensorType())
-                            , Optional.ofNullable(originalMessage)
-                            );
+      MetronError error = new MetronError()
+              .withErrorType(Constants.ErrorType.PARSER_ERROR)
+              .withThrowable(ex)
+              .withSensorType(getSensorType())
+              .addRawMessage(originalMessage);
+      ErrorUtils.handleError(collector, error);
       collector.ack(tuple);
     }
   }
 
-  private boolean isGloballyValid(JSONObject input, List<FieldValidator> validators) {
+  private List<FieldValidator> getFailedValidators(JSONObject input, List<FieldValidator> validators) {
+    List<FieldValidator> failedValidators = new ArrayList<>();
     for(FieldValidator validator : validators) {
       if(!validator.isValid(input, getConfigurations().getGlobalConfig(), stellarContext)) {
-        return false;
+        failedValidators.add(validator);
       }
     }
-    return true;
+    return failedValidators;
   }
 
   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
-    declarer.declareStream(Constants.INVALID_STREAM, new Fields("message"));
     declarer.declareStream(Constants.ERROR_STREAM, new Fields("message"));
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java
index 8eb0656..ef7288b 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java
@@ -18,23 +18,27 @@
 
 package org.apache.metron.parsers.bolt;
 
+import org.apache.metron.common.Constants;
+import org.apache.metron.common.configuration.ParserConfigurations;
+import org.apache.metron.common.error.MetronError;
+import org.apache.metron.common.message.MessageGetStrategy;
+import org.apache.metron.common.message.MessageGetters;
+import org.apache.metron.common.utils.ErrorUtils;
 import org.apache.storm.task.OutputCollector;
 import org.apache.storm.task.TopologyContext;
 import org.apache.storm.topology.OutputFieldsDeclarer;
 import org.apache.storm.topology.base.BaseRichBolt;
 import org.apache.storm.tuple.Tuple;
-import org.apache.metron.common.Constants;
-import org.apache.metron.common.configuration.ParserConfigurations;
-import org.apache.metron.common.utils.ErrorUtils;
 import org.json.simple.JSONObject;
 
 import java.util.Map;
-import java.util.Optional;
 
 public class WriterBolt extends BaseRichBolt {
   private WriterHandler handler;
   private ParserConfigurations configuration;
   private String sensorType;
+  private Constants.ErrorType errorType = Constants.ErrorType.DEFAULT_ERROR;
+  private transient MessageGetStrategy messageGetStrategy;
   private transient OutputCollector collector;
   public WriterBolt(WriterHandler handler, ParserConfigurations configuration, String sensorType) {
     this.handler = handler;
@@ -42,9 +46,15 @@ public class WriterBolt extends BaseRichBolt {
     this.sensorType = sensorType;
   }
 
+  public WriterBolt withErrorType(Constants.ErrorType errorType) {
+    this.errorType = errorType;
+    return this;
+  }
+
   @Override
   public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
     this.collector = collector;
+    messageGetStrategy = MessageGetters.DEFAULT_JSON_FROM_FIELD.get();
     handler.init(stormConf, collector, configuration);
   }
 
@@ -65,18 +75,18 @@ public class WriterBolt extends BaseRichBolt {
   public void execute(Tuple tuple) {
     JSONObject message = null;
     try {
-      message = (JSONObject)((JSONObject) tuple.getValueByField("message")).clone();
-      handler.write(sensorType, tuple, message, configuration);
+      message = (JSONObject) messageGetStrategy.get(tuple);
+      handler.write(sensorType, tuple, message, configuration, messageGetStrategy);
       if(!handler.handleAck()) {
         collector.ack(tuple);
       }
     } catch (Throwable e) {
-      ErrorUtils.handleError( collector
-                            , e
-                            , Constants.ERROR_STREAM
-                            , Optional.of(sensorType)
-                            , Optional.ofNullable(message)
-                            );
+      MetronError error = new MetronError()
+              .withErrorType(errorType)
+              .withThrowable(e)
+              .withSensorType(sensorType)
+              .addRawMessage(message);
+      ErrorUtils.handleError(collector, error);
       collector.ack(tuple);
     }
   }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterHandler.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterHandler.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterHandler.java
index 38425b5..3273ca7 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterHandler.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/bolt/WriterHandler.java
@@ -18,6 +18,7 @@
 
 package org.apache.metron.parsers.bolt;
 
+import org.apache.metron.common.message.MessageGetStrategy;
 import org.apache.storm.task.OutputCollector;
 import org.apache.storm.tuple.Tuple;
 import org.apache.metron.common.configuration.ParserConfigurations;
@@ -80,11 +81,12 @@ public class WriterHandler implements Serializable {
                    , Tuple tuple
                    , JSONObject message
                    , ParserConfigurations configurations
+                   , MessageGetStrategy messageGetStrategy
                    ) throws Exception {
-    writerComponent.write(sensorType, tuple, message, messageWriter, writerTransformer.apply(configurations));
+    writerComponent.write(sensorType, tuple, message, messageWriter, writerTransformer.apply(configurations), messageGetStrategy);
   }
 
-  public void errorAll(String sensorType, Throwable e) {
-    writerComponent.errorAll(sensorType, e);
+  public void errorAll(String sensorType, Throwable e, MessageGetStrategy messageGetStrategy) {
+    writerComponent.errorAll(sensorType, e, messageGetStrategy);
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
index 5f1927e..aeac33c 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
@@ -57,8 +57,6 @@ public class ParserTopologyBuilder {
    * @param spoutNumTasks            Number of tasks for the spout
    * @param parserParallelism        Parallelism hint for the parser bolt
    * @param parserNumTasks           Number of tasks for the parser bolt
-   * @param invalidWriterParallelism Parallelism hint for the bolt that handles invalid data
-   * @param invalidWriterNumTasks    Number of tasks for the bolt that handles invalid data
    * @param errorWriterParallelism   Parallelism hint for the bolt that handles errors
    * @param errorWriterNumTasks      Number of tasks for the bolt that handles errors
    * @param kafkaSpoutConfig         Configuration options for the kafka spout
@@ -73,8 +71,6 @@ public class ParserTopologyBuilder {
                                       int spoutNumTasks,
                                       int parserParallelism,
                                       int parserNumTasks,
-                                      int invalidWriterParallelism,
-                                      int invalidWriterNumTasks,
                                       int errorWriterParallelism,
                                       int errorWriterNumTasks,
                                       EnumMap<SpoutConfigOptions, Object> kafkaSpoutConfig
@@ -104,14 +100,6 @@ public class ParserTopologyBuilder {
               .shuffleGrouping("parserBolt", Constants.ERROR_STREAM);
     }
 
-    // create the invalid bolt, if needed
-    if (invalidWriterNumTasks > 0) {
-      WriterBolt invalidBolt = createInvalidBolt(brokerUrl, sensorType, configs, parserConfig);
-      builder.setBolt("invalidMessageWriter", invalidBolt, invalidWriterParallelism)
-              .setNumTasks(invalidWriterNumTasks)
-              .shuffleGrouping("parserBolt", Constants.INVALID_STREAM);
-    }
-
     return builder;
   }
 
@@ -162,29 +150,6 @@ public class ParserTopologyBuilder {
   }
 
   /**
-   * Create a bolt that handles invalid messages.
-   *
-   * @param brokerUrl    The Kafka Broker URL
-   * @param sensorType   Type of sensor that is being consumed.
-   * @param configs
-   * @param parserConfig
-   * @return A Storm bolt that handles invalid messages.
-   */
-  private static WriterBolt createInvalidBolt(String brokerUrl, String sensorType, ParserConfigurations configs, SensorParserConfig parserConfig) {
-
-    // create writer - if not configured uses a sensible default
-    AbstractWriter writer = parserConfig.getErrorWriterClassName() == null
-            ? new KafkaWriter(brokerUrl).withTopic(Constants.DEFAULT_PARSER_INVALID_TOPIC).withConfigPrefix("invalid")
-            : ReflectionUtils.createInstance(parserConfig.getWriterClassName());
-    writer.configure(sensorType, new ParserWriterConfiguration(configs));
-
-    // create a writer handler
-    WriterHandler writerHandler = createWriterHandler(writer);
-
-    return new WriterBolt(writerHandler, configs, sensorType);
-  }
-
-  /**
    * Create a bolt that handles error messages.
    *
    * @param brokerUrl    Kafka Broker URL
@@ -197,14 +162,14 @@ public class ParserTopologyBuilder {
 
     // create writer - if not configured uses a sensible default
     AbstractWriter writer = parserConfig.getErrorWriterClassName() == null
-            ? new KafkaWriter(brokerUrl).withTopic(Constants.DEFAULT_PARSER_ERROR_TOPIC).withConfigPrefix("error")
+            ? new KafkaWriter(brokerUrl).withTopic((String) configs.getGlobalConfig().get("parser.error.topic")).withConfigPrefix("error")
             : ReflectionUtils.createInstance(parserConfig.getWriterClassName());
     writer.configure(sensorType, new ParserWriterConfiguration(configs));
 
     // create a writer handler
     WriterHandler writerHandler = createWriterHandler(writer);
 
-    return new WriterBolt(writerHandler, configs, sensorType);
+    return new WriterBolt(writerHandler, configs, sensorType).withErrorType(Constants.ErrorType.PARSER_ERROR);
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
index 5ea561c..2bf484e 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyCLI.java
@@ -302,8 +302,6 @@ public class ParserTopologyCLI {
               spoutNumTasks,
               parserParallelism,
               parserNumTasks,
-              invalidParallelism,
-              invalidNumTasks,
               errorParallelism,
               errorNumTasks,
               spoutConfig

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java
index 6b00998..b3e15b2 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java
@@ -17,8 +17,11 @@
  */
 package org.apache.metron.parsers.bolt;
 
+import org.apache.metron.common.Constants;
 import org.apache.metron.common.configuration.*;
 
+import org.apache.metron.common.error.MetronError;
+import org.apache.metron.test.error.MetronErrorJSONMatcher;
 import org.apache.metron.test.utils.UnitTestHelper;
 import org.apache.storm.task.OutputCollector;
 import org.apache.storm.tuple.Tuple;
@@ -151,6 +154,64 @@ public class ParserBoltTest extends BaseBoltTest {
     verify(parser, times(0)).validate(any());
     verify(writer, times(0)).write(eq(sensorType), any(ParserWriterConfiguration.class), eq(tuple), any());
     verify(outputCollector, times(1)).ack(tuple);
+
+    MetronError error = new MetronError()
+            .withErrorType(Constants.ErrorType.PARSER_ERROR)
+            .withThrowable(new NullPointerException())
+            .withSensorType(sensorType)
+            .addRawMessage(sampleBinary);
+    verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), argThat(new MetronErrorJSONMatcher(error.getJSONObject())));
+  }
+
+  @Test
+  public void testInvalid() throws Exception {
+    String sensorType = "yaf";
+    ParserBolt parserBolt = new ParserBolt("zookeeperUrl", sensorType, parser, new WriterHandler(writer)) {
+      @Override
+      protected ParserConfigurations defaultConfigurations() {
+        return new ParserConfigurations() {
+          @Override
+          public SensorParserConfig getSensorParserConfig(String sensorType) {
+            return new SensorParserConfig() {
+              @Override
+              public Map<String, Object> getParserConfig() {
+                return new HashMap<String, Object>() {{
+                }};
+              }
+
+
+            };
+          }
+        };
+      }
+
+    };
+
+    buildGlobalConfig(parserBolt);
+
+    parserBolt.setCuratorFramework(client);
+    parserBolt.setTreeCache(cache);
+    parserBolt.prepare(new HashMap(), topologyContext, outputCollector);
+    byte[] sampleBinary = "some binary message".getBytes();
+
+    when(tuple.getBinary(0)).thenReturn(sampleBinary);
+    JSONObject parsedMessage = new JSONObject();
+    parsedMessage.put("field", "invalidValue");
+    List<JSONObject> messageList = new ArrayList<>();
+    messageList.add(parsedMessage);
+    when(parser.parseOptional(sampleBinary)).thenReturn(Optional.of(messageList));
+    when(parser.validate(parsedMessage)).thenReturn(true);
+    parserBolt.execute(tuple);
+
+    MetronError error = new MetronError()
+            .withErrorType(Constants.ErrorType.PARSER_INVALID)
+            .withSensorType(sensorType)
+            .withErrorFields(new HashSet<String>() {{ add("field"); }})
+            .addRawMessage(new JSONObject(){{
+              put("field", "invalidValue");
+              put("source.type", "yaf");
+            }});
+    verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), argThat(new MetronErrorJSONMatcher(error.getJSONObject())));
   }
 
   @Test
@@ -531,6 +592,16 @@ public void testImplicitBatchOfOne() throws Exception {
 
   }
 
+  protected void buildGlobalConfig(ParserBolt parserBolt) {
+    HashMap<String, Object> globalConfig = new HashMap<>();
+    Map<String, Object> fieldValidation = new HashMap<>();
+    fieldValidation.put("input", Arrays.asList("field"));
+    fieldValidation.put("validation", "STELLAR");
+    fieldValidation.put("config", new HashMap<String, String>(){{ put("condition", "field != 'invalidValue'"); }});
+    globalConfig.put("fieldValidations", Arrays.asList(fieldValidation));
+    parserBolt.getConfigurations().updateGlobalConfig(globalConfig);
+  }
+
   private static void writeNonBatch(OutputCollector collector, ParserBolt bolt, Tuple t) {
     bolt.execute(t);
   }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java
index 4693829..4511b55 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java
@@ -17,22 +17,23 @@
  */
 
 package org.apache.metron.parsers.bolt;
+
 import org.apache.log4j.Level;
+import org.apache.metron.common.Constants;
 import org.apache.metron.common.configuration.IndexingConfigurations;
-import org.apache.metron.test.utils.UnitTestHelper;
-import org.apache.metron.writer.BulkWriterComponent;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.tuple.Tuple;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
 import org.apache.metron.common.configuration.ParserConfigurations;
 import org.apache.metron.common.configuration.SensorParserConfig;
-import org.apache.metron.common.configuration.writer.ParserWriterConfiguration;
+import org.apache.metron.common.error.MetronError;
 import org.apache.metron.common.writer.BulkMessageWriter;
-import org.apache.metron.common.writer.MessageWriter;
 import org.apache.metron.common.writer.BulkWriterResponse;
+import org.apache.metron.common.writer.MessageWriter;
 import org.apache.metron.test.bolt.BaseBoltTest;
+import org.apache.metron.test.error.MetronErrorJSONMatcher;
+import org.apache.metron.test.utils.UnitTestHelper;
+import org.apache.metron.writer.BulkWriterComponent;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.tuple.Tuple;
 import org.json.simple.JSONObject;
 import org.junit.Test;
 import org.mockito.Mock;
@@ -44,7 +45,12 @@ import java.util.Map;
 
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.argThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class WriterBoltTest extends BaseBoltTest{
   @Mock
@@ -143,16 +149,24 @@ public class WriterBoltTest extends BaseBoltTest{
     ParserConfigurations configurations = getConfigurations(1);
     String sensorType = "test";
     Tuple t = mock(Tuple.class);
+    when(t.toString()).thenReturn("tuple");
     when(t.getValueByField(eq("message"))).thenReturn(new JSONObject());
     WriterBolt bolt = new WriterBolt(new WriterHandler(writer), configurations, sensorType);
     bolt.prepare(new HashMap(), topologyContext, outputCollector);
-    doThrow(new Exception()).when(writer).write(any(), any(), any(), any());
+    doThrow(new Exception("write error")).when(writer).write(any(), any(), any(), any());
     verify(writer, times(1)).init();
     bolt.execute(t);
     verify(outputCollector, times(1)).ack(t);
     verify(writer, times(1)).write(eq(sensorType), any(), any(), any());
     verify(outputCollector, times(1)).reportError(any());
     verify(outputCollector, times(0)).fail(any());
+
+    MetronError error = new MetronError()
+            .withErrorType(Constants.ErrorType.DEFAULT_ERROR)
+            .withThrowable(new IllegalStateException("Unhandled bulk errors in response: {java.lang.Exception: write error=[tuple]}"))
+            .withSensorType(sensorType)
+            .addRawMessage(new JSONObject());
+    verify(outputCollector, times(1)).emit(eq(Constants.ERROR_STREAM), argThat(new MetronErrorJSONMatcher(error.getJSONObject())));
   }
 
   @Test

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserIntegrationTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserIntegrationTest.java
index 4ba1c43..a170a2c 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserIntegrationTest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/ParserIntegrationTest.java
@@ -37,6 +37,7 @@ import javax.annotation.Nullable;
 import java.util.*;
 
 public abstract class ParserIntegrationTest extends BaseIntegrationTest {
+  protected static final String ERROR_TOPIC = "parser_error";
   protected List<byte[]> inputMessages;
   @Test
   public void test() throws Exception {
@@ -47,8 +48,7 @@ public abstract class ParserIntegrationTest extends BaseIntegrationTest {
     final KafkaComponent kafkaComponent = getKafkaComponent(topologyProperties, new ArrayList<KafkaComponent.Topic>() {{
       add(new KafkaComponent.Topic(sensorType, 1));
       add(new KafkaComponent.Topic(Constants.ENRICHMENT_TOPIC, 1));
-      add(new KafkaComponent.Topic(Constants.INVALID_STREAM,1));
-      add(new KafkaComponent.Topic(Constants.ERROR_STREAM,1));
+      add(new KafkaComponent.Topic(ERROR_TOPIC,1));
     }});
     topologyProperties.setProperty("kafka.broker", kafkaComponent.getBrokerList());
 
@@ -115,13 +115,12 @@ public abstract class ParserIntegrationTest extends BaseIntegrationTest {
     return new KafkaProcessor<>()
             .withKafkaComponentName("kafka")
             .withReadTopic(Constants.ENRICHMENT_TOPIC)
-            .withErrorTopic(Constants.ERROR_STREAM)
-            .withInvalidTopic(Constants.INVALID_STREAM)
+            .withErrorTopic(ERROR_TOPIC)
             .withValidateReadMessages(new Function<KafkaMessageSet, Boolean>() {
               @Nullable
               @Override
               public Boolean apply(@Nullable KafkaMessageSet messageSet) {
-                return (messageSet.getMessages().size() + messageSet.getErrors().size() + messageSet.getInvalids().size()) == inputMessages.size();
+                return (messageSet.getMessages().size() + messageSet.getErrors().size() == inputMessages.size());
               }
             })
             .withProvideResult(new Function<KafkaMessageSet,List<byte[]>>(){

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java
index 80b6ebd..73d3827 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/parsers/integration/components/ParserTopologyComponent.java
@@ -34,6 +34,7 @@ public class ParserTopologyComponent implements InMemoryComponent {
   private Properties topologyProperties;
   private String brokerUrl;
   private String sensorType;
+  private SpoutConfig.Offset offset = SpoutConfig.Offset.BEGINNING;
   private LocalCluster stormCluster;
 
   public static class Builder {
@@ -64,15 +65,17 @@ public class ParserTopologyComponent implements InMemoryComponent {
     this.sensorType = sensorType;
   }
 
+  public void setOffset(SpoutConfig.Offset offset) {
+    this.offset = offset;
+  }
+
   @Override
   public void start() throws UnableToStartException {
     try {
       TopologyBuilder topologyBuilder = ParserTopologyBuilder.build(topologyProperties.getProperty("kafka.zk")
                                                                    , brokerUrl
                                                                    , sensorType
-                                                                   , SpoutConfig.Offset.BEGINNING
-                                                                   , 1
-                                                                   , 1
+                                                                   , offset
                                                                    , 1
                                                                    , 1
                                                                    , 1

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java
index f37b1fc..7476bcf 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/WriterBoltIntegrationTest.java
@@ -15,8 +15,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.metron.writers.integration;
+
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.google.common.base.Function;
 import com.google.common.collect.Iterables;
@@ -37,6 +37,8 @@ import org.apache.metron.parsers.csv.CSVParser;
 import org.apache.metron.parsers.integration.components.ParserTopologyComponent;
 import org.apache.metron.test.utils.UnitTestHelper;
 import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.json.simple.parser.ParseException;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -45,6 +47,8 @@ import java.io.IOException;
 import java.util.*;
 
 public class WriterBoltIntegrationTest extends BaseIntegrationTest {
+  private static final String ERROR_TOPIC = "parser_error";
+
   public static class MockValidator implements FieldValidation{
 
     @Override
@@ -65,7 +69,8 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
         {
           "validation" : "org.apache.metron.writers.integration.WriterBoltIntegrationTest$MockValidator"
         }
-                         ]
+                         ],
+   "parser.error.topic":"parser_error"
    }
     */
   @Multiline
@@ -88,7 +93,7 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
   public static String parserConfig;
 
   @Test
-  public void test() throws UnableToStartException, IOException {
+  public void test() throws UnableToStartException, IOException, ParseException {
     UnitTestHelper.setLog4jLevel(CSVParser.class, org.apache.log4j.Level.FATAL);
     final String sensorType = "dummy";
     final List<byte[]> inputMessages = new ArrayList<byte[]>() {{
@@ -100,8 +105,7 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
     final ZKServerComponent zkServerComponent = getZKServerComponent(topologyProperties);
     final KafkaComponent kafkaComponent = getKafkaComponent(topologyProperties, new ArrayList<KafkaComponent.Topic>() {{
       add(new KafkaComponent.Topic(sensorType, 1));
-      add(new KafkaComponent.Topic(Constants.DEFAULT_PARSER_ERROR_TOPIC, 1));
-      add(new KafkaComponent.Topic(Constants.DEFAULT_PARSER_INVALID_TOPIC, 1));
+      add(new KafkaComponent.Topic(ERROR_TOPIC, 1));
       add(new KafkaComponent.Topic(Constants.ENRICHMENT_TOPIC, 1));
     }});
     topologyProperties.setProperty("kafka.broker", kafkaComponent.getBrokerList());
@@ -131,17 +135,20 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
       kafkaComponent.writeMessages(sensorType, inputMessages);
       ProcessorResult<Map<String, List<JSONObject>>> result = runner.process(getProcessor());
       Map<String,List<JSONObject>> outputMessages = result.getResult();
-      Assert.assertEquals(3, outputMessages.size());
+      Assert.assertEquals(2, outputMessages.size());
       Assert.assertEquals(1, outputMessages.get(Constants.ENRICHMENT_TOPIC).size());
       Assert.assertEquals("valid", outputMessages.get(Constants.ENRICHMENT_TOPIC).get(0).get("action"));
-      Assert.assertEquals(1, outputMessages.get(Constants.DEFAULT_PARSER_ERROR_TOPIC).size());
-      Assert.assertEquals("error", outputMessages.get(Constants.DEFAULT_PARSER_ERROR_TOPIC).get(0).get("rawMessage"));
-      Assert.assertTrue(Arrays.equals(listToBytes(outputMessages.get(Constants.DEFAULT_PARSER_ERROR_TOPIC).get(0).get("rawMessage_bytes"))
-                                     , "error".getBytes()
-                                     )
-                      );
-      Assert.assertEquals(1, outputMessages.get(Constants.DEFAULT_PARSER_INVALID_TOPIC).size());
-      Assert.assertEquals("invalid", outputMessages.get(Constants.DEFAULT_PARSER_INVALID_TOPIC).get(0).get("action"));
+      Assert.assertEquals(2, outputMessages.get(ERROR_TOPIC).size());
+      JSONObject invalidMessage = outputMessages.get(ERROR_TOPIC).get(0);
+      Assert.assertEquals(Constants.ErrorType.PARSER_INVALID.getType(), invalidMessage.get(Constants.ErrorFields.ERROR_TYPE.getName()));
+      JSONObject rawMessage = JSONUtils.INSTANCE.load((String) invalidMessage.get(Constants.ErrorFields.RAW_MESSAGE.getName()), JSONObject.class);
+      Assert.assertEquals("foo", rawMessage.get("dummy"));
+      Assert.assertEquals("invalid", rawMessage.get("action"));
+      JSONObject errorMessage = outputMessages.get(ERROR_TOPIC).get(1);
+      Assert.assertEquals(Constants.ErrorType.PARSER_ERROR.getType(), errorMessage.get(Constants.ErrorFields.ERROR_TYPE.getName()));
+      Assert.assertEquals("error", errorMessage.get(Constants.ErrorFields.RAW_MESSAGE.getName()));
+      // It's unclear if we need a rawMessageBytes field so commenting out for now
+      //Assert.assertTrue(Arrays.equals(listToBytes(errorMessage.get(Constants.ErrorFields.RAW_MESSAGE_BYTES.getName())), "error".getBytes()));
     }
     finally {
       if(runner != null) {
@@ -182,13 +189,12 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
     return new KafkaProcessor<>()
             .withKafkaComponentName("kafka")
             .withReadTopic(Constants.ENRICHMENT_TOPIC)
-            .withErrorTopic(Constants.DEFAULT_PARSER_ERROR_TOPIC)
-            .withInvalidTopic(Constants.DEFAULT_PARSER_INVALID_TOPIC)
+            .withErrorTopic(ERROR_TOPIC)
             .withValidateReadMessages(new Function<KafkaMessageSet, Boolean>() {
               @Nullable
               @Override
               public Boolean apply(@Nullable KafkaMessageSet messageSet) {
-                return (messageSet.getMessages().size() == 1) && (messageSet.getErrors().size() == 1) && (messageSet.getInvalids().size() ==1);
+                return (messageSet.getMessages().size() == 1) && (messageSet.getErrors().size() == 2);
               }
             })
             .withProvideResult(new Function<KafkaMessageSet,Map<String,List<JSONObject>>>(){
@@ -197,8 +203,7 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
               public Map<String,List<JSONObject>> apply(@Nullable KafkaMessageSet messageSet) {
                 return new HashMap<String, List<JSONObject>>() {{
                   put(Constants.ENRICHMENT_TOPIC, loadMessages(messageSet.getMessages()));
-                  put(Constants.DEFAULT_PARSER_ERROR_TOPIC, loadMessages(messageSet.getErrors()));
-                  put(Constants.DEFAULT_PARSER_INVALID_TOPIC, loadMessages(messageSet.getInvalids()));
+                  put(ERROR_TOPIC, loadMessages(messageSet.getErrors()));
                 }};
               }
             });

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-solr/src/main/config/solr.properties
----------------------------------------------------------------------
diff --git a/metron-platform/metron-solr/src/main/config/solr.properties b/metron-platform/metron-solr/src/main/config/solr.properties
index 8489d8e..d8dd25f 100644
--- a/metron-platform/metron-solr/src/main/config/solr.properties
+++ b/metron-platform/metron-solr/src/main/config/solr.properties
@@ -26,7 +26,7 @@ kafka.start=WHERE_I_LEFT_OFF
 
 ##### Indexing #####
 index.input.topic=indexing
-index.error.topic=indexing_error
+index.error.topic=indexing
 writer.class.name=org.apache.metron.solr.writer.SolrWriter
 
 ##### Metrics #####

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/integration/SolrIndexingIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/integration/SolrIndexingIntegrationTest.java b/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/integration/SolrIndexingIntegrationTest.java
index abf3a8a..c209ef3 100644
--- a/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/integration/SolrIndexingIntegrationTest.java
+++ b/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/integration/SolrIndexingIntegrationTest.java
@@ -18,14 +18,17 @@
 package org.apache.metron.solr.integration;
 
 import com.google.common.base.Function;
-import org.apache.metron.common.Constants;
 import org.apache.metron.common.configuration.Configurations;
 import org.apache.metron.common.configuration.ConfigurationsUtils;
 import org.apache.metron.common.interfaces.FieldNameConverter;
 import org.apache.metron.common.utils.JSONUtils;
 import org.apache.metron.enrichment.integration.utils.SampleUtil;
 import org.apache.metron.indexing.integration.IndexingIntegrationTest;
-import org.apache.metron.integration.*;
+import org.apache.metron.integration.ComponentRunner;
+import org.apache.metron.integration.InMemoryComponent;
+import org.apache.metron.integration.Processor;
+import org.apache.metron.integration.ProcessorResult;
+import org.apache.metron.integration.ReadinessState;
 import org.apache.metron.integration.components.KafkaComponent;
 import org.apache.metron.integration.components.ZKServerComponent;
 import org.apache.metron.solr.integration.components.SolrComponent;
@@ -89,7 +92,7 @@ public class SolrIndexingIntegrationTest extends IndexingIntegrationTest {
             throw new IllegalStateException("Unable to retrieve indexed documents.", e);
           }
           if (docs.size() < inputMessages.size() || docs.size() != docsFromDisk.size()) {
-            errors = kafkaComponent.readMessages(Constants.INDEXING_ERROR_TOPIC);
+            errors = kafkaComponent.readMessages(ERROR_TOPIC);
             if(errors.size() > 0){
               return ReadinessState.READY;
             }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/error/MetronErrorJSONMatcher.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/error/MetronErrorJSONMatcher.java b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/error/MetronErrorJSONMatcher.java
new file mode 100644
index 0000000..ad24283
--- /dev/null
+++ b/metron-platform/metron-test-utilities/src/main/java/org/apache/metron/test/error/MetronErrorJSONMatcher.java
@@ -0,0 +1,42 @@
+/**
+ * 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.test.error;
+
+import org.apache.storm.tuple.Values;
+import org.json.simple.JSONObject;
+import org.mockito.ArgumentMatcher;
+
+public class MetronErrorJSONMatcher extends ArgumentMatcher<Values> {
+
+  private JSONObject expected;
+
+  public MetronErrorJSONMatcher(JSONObject expected) {
+    this.expected = expected;
+  }
+
+  @Override
+  public boolean matches(Object o) {
+    Values values = (Values) o;
+    JSONObject actual = (JSONObject) values.get(0);
+    actual.remove("timestamp");
+    expected.remove("timestamp");
+    actual.remove("stack");
+    expected.remove("stack");
+    return actual.equals(expected);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
index 124ffd3..0a9e514 100644
--- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
+++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/BulkWriterComponent.java
@@ -18,18 +18,26 @@
 
 package org.apache.metron.writer;
 
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.tuple.Tuple;
 import com.google.common.collect.Iterables;
 import org.apache.metron.common.Constants;
 import org.apache.metron.common.configuration.writer.WriterConfiguration;
+import org.apache.metron.common.error.MetronError;
+import org.apache.metron.common.message.MessageGetStrategy;
+import org.apache.metron.common.utils.ErrorUtils;
 import org.apache.metron.common.writer.BulkMessageWriter;
 import org.apache.metron.common.writer.BulkWriterResponse;
-import org.apache.metron.common.utils.ErrorUtils;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.tuple.Tuple;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 
 public class BulkWriterComponent<MESSAGE_T> {
   public static final Logger LOG = LoggerFactory
@@ -60,18 +68,23 @@ public class BulkWriterComponent<MESSAGE_T> {
     commit(response.getSuccesses());
   }
 
-  public void error(Throwable e, Iterable<Tuple> tuples) {
+  public void error(String sensorType, Throwable e, Iterable<Tuple> tuples, MessageGetStrategy messageGetStrategy) {
     tuples.forEach(t -> collector.ack(t));
+    MetronError error = new MetronError()
+            .withSensorType(sensorType)
+            .withErrorType(Constants.ErrorType.INDEXING_ERROR)
+            .withThrowable(e);
     if(!Iterables.isEmpty(tuples)) {
       LOG.error("Failing " + Iterables.size(tuples) + " tuples", e);
-      ErrorUtils.handleError(collector, e, Constants.ERROR_STREAM);
     }
+    tuples.forEach(t -> error.addRawMessage(messageGetStrategy.get(t)));
+    ErrorUtils.handleError(collector, error);
   }
 
-  public void error(BulkWriterResponse errors) {
+  public void error(String sensorType, BulkWriterResponse errors, MessageGetStrategy messageGetStrategy) {
     Map<Throwable, Collection<Tuple>> errorMap = errors.getErrors();
     for(Map.Entry<Throwable, Collection<Tuple>> entry : errorMap.entrySet()) {
-      error(entry.getKey(), entry.getValue());
+      error(sensorType, entry.getKey(), entry.getValue(), messageGetStrategy);
     }
   }
 
@@ -80,24 +93,25 @@ public class BulkWriterComponent<MESSAGE_T> {
   }
 
 
-  public void errorAll(Throwable e) {
-    for(Map.Entry<String, Collection<Tuple>> kv : sensorTupleMap.entrySet()) {
-      error(e, kv.getValue());
-      sensorTupleMap.remove(kv.getKey());
-      sensorMessageMap.remove(kv.getKey());
+  public void errorAll(Throwable e, MessageGetStrategy messageGetStrategy) {
+    for(String key : new HashSet<>(sensorTupleMap.keySet())) {
+      errorAll(key, e, messageGetStrategy);
     }
   }
 
-  public void errorAll(String sensorType, Throwable e) {
-    error(e, Optional.ofNullable(sensorTupleMap.get(sensorType)).orElse(new ArrayList<>()));
+  public void errorAll(String sensorType, Throwable e, MessageGetStrategy messageGetStrategy) {
+    Collection<Tuple> tuples = Optional.ofNullable(sensorTupleMap.get(sensorType)).orElse(new ArrayList<>());
+    error(sensorType, e, tuples, messageGetStrategy);
     sensorTupleMap.remove(sensorType);
     sensorMessageMap.remove(sensorType);
   }
+
   public void write( String sensorType
                    , Tuple tuple
                    , MESSAGE_T message
                    , BulkMessageWriter<MESSAGE_T> bulkMessageWriter
                    , WriterConfiguration configurations
+                   , MessageGetStrategy messageGetStrategy
                    ) throws Exception
   {
     if(!configurations.isEnabled(sensorType)) {
@@ -129,13 +143,13 @@ public class BulkWriterComponent<MESSAGE_T> {
         }
 
         if(handleError) {
-          error(response);
+          error(sensorType, response, messageGetStrategy);
         } else if (response.hasErrors()) {
           throw new IllegalStateException("Unhandled bulk errors in response: " + response.getErrors());
         }
       } catch (Throwable e) {
         if(handleError) {
-          error(e, tupleList);
+          error(sensorType, e, tupleList, messageGetStrategy);
         }
         else {
           throw e;

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java
index 66c4c73..085ca5c 100644
--- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java
+++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/bolt/BulkMessageWriterBolt.java
@@ -17,27 +17,27 @@
  */
 package org.apache.metron.writer.bolt;
 
-import org.apache.metron.common.bolt.ConfiguredIndexingBolt;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
 import org.apache.metron.common.Constants;
+import org.apache.metron.common.bolt.ConfiguredIndexingBolt;
 import org.apache.metron.common.configuration.writer.IndexingWriterConfiguration;
 import org.apache.metron.common.configuration.writer.WriterConfiguration;
-import org.apache.metron.common.writer.MessageWriter;
+import org.apache.metron.common.message.MessageGetStrategy;
+import org.apache.metron.common.message.MessageGetters;
 import org.apache.metron.common.utils.MessageUtils;
 import org.apache.metron.common.writer.BulkMessageWriter;
+import org.apache.metron.common.writer.MessageWriter;
 import org.apache.metron.writer.BulkWriterComponent;
 import org.apache.metron.writer.WriterToBulkWriter;
-import org.apache.metron.writer.message.MessageGetter;
-import org.apache.metron.writer.message.MessageGetters;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
 import org.json.simple.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.*;
+import java.util.Map;
 import java.util.function.Function;
 
 public class BulkMessageWriterBolt extends ConfiguredIndexingBolt {
@@ -46,8 +46,9 @@ public class BulkMessageWriterBolt extends ConfiguredIndexingBolt {
           .getLogger(BulkMessageWriterBolt.class);
   private BulkMessageWriter<JSONObject> bulkMessageWriter;
   private BulkWriterComponent<JSONObject> writerComponent;
-  private String messageGetterStr = MessageGetters.NAMED.name();
-  private transient MessageGetter messageGetter = null;
+  private String messageGetStrategyType = MessageGetters.DEFAULT_JSON_FROM_FIELD.name();
+  private String messageGetField;
+  private transient MessageGetStrategy messageGetStrategy;
   private transient OutputCollector collector;
   private transient Function<WriterConfiguration, WriterConfiguration> configurationTransformation;
   public BulkMessageWriterBolt(String zookeeperUrl) {
@@ -64,8 +65,13 @@ public class BulkMessageWriterBolt extends ConfiguredIndexingBolt {
     return this;
   }
 
-  public BulkMessageWriterBolt withMessageGetter(String messageGetter) {
-    this.messageGetterStr = messageGetter;
+  public BulkMessageWriterBolt withMessageGetter(String messageGetStrategyType) {
+    this.messageGetStrategyType = messageGetStrategyType;
+    return this;
+  }
+
+  public BulkMessageWriterBolt withMessageGetterField(String messageGetField) {
+    this.messageGetField = messageGetField;
     return this;
   }
 
@@ -74,7 +80,11 @@ public class BulkMessageWriterBolt extends ConfiguredIndexingBolt {
     this.writerComponent = new BulkWriterComponent<>(collector);
     this.collector = collector;
     super.prepare(stormConf, context, collector);
-    messageGetter = MessageGetters.valueOf(messageGetterStr);
+    if (messageGetField != null) {
+      messageGetStrategy = MessageGetters.valueOf(messageGetStrategyType).get(messageGetField);
+    } else {
+      messageGetStrategy = MessageGetters.valueOf(messageGetStrategyType).get();
+    }
     if(bulkMessageWriter instanceof WriterToBulkWriter) {
       configurationTransformation = WriterToBulkWriter.TRANSFORMATION;
     }
@@ -93,7 +103,7 @@ public class BulkMessageWriterBolt extends ConfiguredIndexingBolt {
   @SuppressWarnings("unchecked")
   @Override
   public void execute(Tuple tuple) {
-    JSONObject message = messageGetter.getMessage(tuple);
+    JSONObject message = (JSONObject) messageGetStrategy.get(tuple);
     String sensorType = MessageUtils.getSensorType(message);
     try
     {
@@ -107,6 +117,7 @@ public class BulkMessageWriterBolt extends ConfiguredIndexingBolt {
                            , message
                            , bulkMessageWriter
                            , writerConfiguration
+                           , messageGetStrategy
                            );
       LOG.trace("Writing enrichment message: {}", message);
     }

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/134a2331/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/message/MessageGetter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/message/MessageGetter.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/message/MessageGetter.java
deleted file mode 100644
index 99c825f..0000000
--- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/message/MessageGetter.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/**
- * 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.writer.message;
-
-import org.apache.storm.tuple.Tuple;
-import org.json.simple.JSONObject;
-
-public interface MessageGetter {
-  JSONObject getMessage(Tuple t);
-}



Mime
View raw message