You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by mm...@apache.org on 2016/10/21 18:33:25 UTC
incubator-metron git commit: METRON-502: Make the
ParserIntegrationTest more clear on errors (ottobackwards via mmiklavc)
closes apache/incubator-metron#311
Repository: incubator-metron
Updated Branches:
refs/heads/master 4de80b35e -> d6d3d4572
METRON-502: Make the ParserIntegrationTest more clear on errors (ottobackwards via mmiklavc) closes apache/incubator-metron#311
Project: http://git-wip-us.apache.org/repos/asf/incubator-metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-metron/commit/d6d3d457
Tree: http://git-wip-us.apache.org/repos/asf/incubator-metron/tree/d6d3d457
Diff: http://git-wip-us.apache.org/repos/asf/incubator-metron/diff/d6d3d457
Branch: refs/heads/master
Commit: d6d3d45725d7371e560beea42a2788237aab2052
Parents: 4de80b3
Author: ottobackwards <ot...@gmail.com>
Authored: Fri Oct 21 14:32:46 2016 -0400
Committer: Michael Miklavcic <mi...@gmail.com>
Committed: Fri Oct 21 14:32:46 2016 -0400
----------------------------------------------------------------------
.../ElasticsearchIndexingIntegrationTest.java | 14 ++-
.../integration/EnrichmentIntegrationTest.java | 62 +++++++++----
.../integration/IndexingIntegrationTest.java | 35 ++++++--
.../metron/integration/ComponentRunner.java | 2 +-
.../apache/metron/integration/Processor.java | 2 +-
.../metron/integration/ProcessorResult.java | 93 ++++++++++++++++++++
.../integration/ParserIntegrationTest.java | 86 ++++++++++--------
...pleHbaseEnrichmentWriterIntegrationTest.java | 9 +-
.../integration/WriterBoltIntegrationTest.java | 8 +-
.../PcapTopologyIntegrationTest.java | 3 +-
.../SolrIndexingIntegrationTest.java | 13 ++-
11 files changed, 251 insertions(+), 76 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchIndexingIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchIndexingIntegrationTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchIndexingIntegrationTest.java
index b27c545..a2629f8 100644
--- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchIndexingIntegrationTest.java
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchIndexingIntegrationTest.java
@@ -17,13 +17,16 @@
*/
package org.apache.metron.elasticsearch.integration;
+import org.apache.metron.common.Constants;
import org.apache.metron.common.interfaces.FieldNameConverter;
import org.apache.metron.elasticsearch.writer.ElasticsearchFieldNameConverter;
import org.apache.metron.indexing.integration.IndexingIntegrationTest;
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.KafkaWithZKComponent;
import org.apache.metron.elasticsearch.integration.components.ElasticSearchComponent;
import java.io.File;
@@ -58,8 +61,10 @@ public class ElasticsearchIndexingIntegrationTest extends IndexingIntegrationTes
public Processor<List<Map<String, Object>>> getProcessor(final List<byte[]> inputMessages) {
return new Processor<List<Map<String, Object>>>() {
List<Map<String, Object>> docs = null;
+ List<byte[]> errors = null;
public ReadinessState process(ComponentRunner runner) {
ElasticSearchComponent elasticSearchComponent = runner.getComponent("search", ElasticSearchComponent.class);
+ KafkaWithZKComponent kafkaWithZKComponent = runner.getComponent("kafka", KafkaWithZKComponent.class);
if (elasticSearchComponent.hasIndex(index)) {
List<Map<String, Object>> docsFromDisk;
try {
@@ -70,6 +75,10 @@ public class ElasticsearchIndexingIntegrationTest extends IndexingIntegrationTes
throw new IllegalStateException("Unable to retrieve indexed documents.", e);
}
if (docs.size() < inputMessages.size() || docs.size() != docsFromDisk.size()) {
+ errors = kafkaWithZKComponent.readMessages(Constants.INDEXING_ERROR_TOPIC);
+ if(errors.size() > 0){
+ return ReadinessState.READY;
+ }
return ReadinessState.NOT_READY;
} else {
return ReadinessState.READY;
@@ -79,8 +88,9 @@ public class ElasticsearchIndexingIntegrationTest extends IndexingIntegrationTes
}
}
- public List<Map<String, Object>> getResult() {
- return docs;
+ public ProcessorResult<List<Map<String, Object>>> getResult() {
+ ProcessorResult.Builder<List<Map<String,Object>>> builder = new ProcessorResult.Builder();
+ return builder.withResult(docs).withProcessErrors(errors).build();
}
};
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/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 48ae918..9f5c29c 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
@@ -95,7 +95,7 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
"{\"ip\":\"10.1.128.237\", \"local\":\"UNKNOWN\", \"type\":\"unknown\", \"asset_value\" : \"important\"},\n" +
"{\"ip\":\"10.60.10.254\", \"local\":\"YES\", \"type\":\"printer\", \"asset_value\" : \"important\"},\n" +
"{\"ip\":\"10.0.2.15\", \"local\":\"YES\", \"type\":\"printer\", \"asset_value\" : \"important\"}]");
- setProperty("hbase.provider.impl","" + Provider.class.getName());
+ setProperty("hbase.provider.impl", "" + Provider.class.getName());
setProperty("threat.intel.tracker.table", trackerHBaseTableName);
setProperty("threat.intel.tracker.cf", cf);
setProperty("threat.intel.simple.hbase.table", threatIntelTableName);
@@ -122,21 +122,20 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
ConfigUploadComponent configUploadComponent = new ConfigUploadComponent()
.withTopologyProperties(topologyProperties)
.withGlobalConfig(globalConfigStr)
- .withEnrichmentConfigsPath(TestConstants.SAMPLE_CONFIG_PATH)
- ;
+ .withEnrichmentConfigsPath(TestConstants.SAMPLE_CONFIG_PATH);
//create MockHBaseTables
- final MockHTable trackerTable = (MockHTable)MockHTable.Provider.addToCache(trackerHBaseTableName, cf);
- final MockHTable threatIntelTable = (MockHTable)MockHTable.Provider.addToCache(threatIntelTableName, cf);
- EnrichmentHelper.INSTANCE.load(threatIntelTable, cf, new ArrayList<LookupKV<EnrichmentKey, EnrichmentValue>>(){{
+ final MockHTable trackerTable = (MockHTable) MockHTable.Provider.addToCache(trackerHBaseTableName, cf);
+ final MockHTable threatIntelTable = (MockHTable) MockHTable.Provider.addToCache(threatIntelTableName, cf);
+ EnrichmentHelper.INSTANCE.load(threatIntelTable, cf, new ArrayList<LookupKV<EnrichmentKey, EnrichmentValue>>() {{
add(new LookupKV<>(new EnrichmentKey(MALICIOUS_IP_TYPE, "10.0.2.3"), new EnrichmentValue(new HashMap<>())));
}});
- final MockHTable enrichmentTable = (MockHTable)MockHTable.Provider.addToCache(enrichmentsTableName, cf);
- EnrichmentHelper.INSTANCE.load(enrichmentTable, cf, new ArrayList<LookupKV<EnrichmentKey, EnrichmentValue>>(){{
+ final MockHTable enrichmentTable = (MockHTable) MockHTable.Provider.addToCache(enrichmentsTableName, cf);
+ EnrichmentHelper.INSTANCE.load(enrichmentTable, cf, new ArrayList<LookupKV<EnrichmentKey, EnrichmentValue>>() {{
add(new LookupKV<>(new EnrichmentKey(PLAYFUL_CLASSIFICATION_TYPE, "10.0.2.3")
- , new EnrichmentValue(PLAYFUL_ENRICHMENT )
- )
- );
+ , new EnrichmentValue(PLAYFUL_ENRICHMENT)
+ )
+ );
}});
FluxTopologyComponent fluxComponent = new FluxTopologyComponent.Builder()
@@ -160,16 +159,31 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
fluxComponent.submitTopology();
kafkaComponent.writeMessages(Constants.ENRICHMENT_TOPIC, inputMessages);
- List<Map<String, Object>> docs = runner.process(getProcessor(inputMessages));
- Assert.assertEquals(inputMessages.size(), docs.size());
- List<Map<String, Object>> cleanedDocs = docs;
- validateAll(cleanedDocs);
- }
- finally {
+ ProcessorResult<List<Map<String, Object>>> result = runner.process(getProcessor(inputMessages));
+ List<Map<String, Object>> docs = result.getResult();
+ if (result.failed()){
+ StringBuffer buffer = new StringBuffer();
+ result.getBadResults(buffer);
+ buffer.append(String.format("%d Valid Messages Processed", docs.size())).append("\n");
+ dumpParsedMessages(docs,buffer);
+ Assert.fail(buffer.toString());
+ } else {
+ Assert.assertEquals(inputMessages.size(), docs.size());
+ List<Map<String, Object>> cleanedDocs = docs;
+ validateAll(cleanedDocs);
+ }
+ } finally {
runner.stop();
}
}
+ public void dumpParsedMessages(List<Map<String,Object>> outputMessages, StringBuffer buffer) {
+ for (Map<String,Object> map : outputMessages) {
+ for( String json : map.keySet()) {
+ buffer.append(json).append("\n");
+ }
+ }
+ }
public static void validateAll(List<Map<String, Object>> docs) {
@@ -425,6 +439,8 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
public Processor<List<Map<String, Object>>> getProcessor(List<byte[]> inputMessages) {
return new Processor<List<Map<String, Object>>>() {
List<Map<String, Object>> docs = null;
+ List<byte[]> errors = null;
+ List<byte[]> invalids = null;
public ReadinessState process(ComponentRunner runner) {
KafkaWithZKComponent kafkaComponent = runner.getComponent("kafka", KafkaWithZKComponent.class);
@@ -440,12 +456,20 @@ public class EnrichmentIntegrationTest extends BaseIntegrationTest {
}
return ReadinessState.READY;
} else {
+ errors = kafkaComponent.readMessages(Constants.ERROR_STREAM);
+ invalids = kafkaComponent.readMessages(Constants.INVALID_STREAM);
+ if(errors.size() > 0 || invalids.size() > 0) {
+ messages = messages;
+ return ReadinessState.READY;
+ }
return ReadinessState.NOT_READY;
}
}
- public List<Map<String, Object>> getResult() {
- return docs;
+ public ProcessorResult<List<Map<String, Object>>> getResult()
+ {
+ ProcessorResult.Builder<List<Map<String,Object>>> builder = new ProcessorResult.Builder();
+ return builder.withResult(docs).withProcessErrors(errors).withProcessInvalids(invalids).build();
}
};
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/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 2c85308..99c4694 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
@@ -31,8 +31,9 @@ import org.apache.metron.integration.BaseIntegrationTest;
import org.apache.metron.integration.ComponentRunner;
import org.apache.metron.integration.InMemoryComponent;
import org.apache.metron.integration.Processor;
-import org.apache.metron.integration.components.FluxTopologyComponent;
+import org.apache.metron.integration.ProcessorResult;
import org.apache.metron.integration.components.KafkaWithZKComponent;
+import org.apache.metron.integration.components.FluxTopologyComponent;
import org.apache.metron.integration.utils.TestUtils;
import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy;
import org.junit.Assert;
@@ -161,6 +162,7 @@ public abstract class IndexingIntegrationTest extends BaseIntegrationTest {
fluxComponent.submitTopology();
kafkaComponent.writeMessages(Constants.INDEXING_TOPIC, inputMessages);
+ StringBuffer buffer = new StringBuffer();
List<Map<String, Object>> docs = cleanDocs(runner.process(getProcessor(inputMessages)));
Assert.assertEquals(docs.size(), inputMessages.size());
//assert that our input docs are equivalent to the output docs, converting the input docs keys based
@@ -175,15 +177,32 @@ public abstract class IndexingIntegrationTest extends BaseIntegrationTest {
}
}
- public List<Map<String, Object>> cleanDocs(List<Map<String, Object>> docs) {
+ public List<Map<String, Object>> cleanDocs(ProcessorResult<List<Map<String, Object>>> result) {
+ List<Map<String,Object>> docs = result.getResult();
+ StringBuffer buffer = new StringBuffer();
+ boolean failed = false;
List<Map<String, Object>> ret = new ArrayList<>();
- for (Map<String, Object> doc : docs) {
- Map<String, Object> msg = new HashMap<>();
- for (Map.Entry<String, Object> kv : doc.entrySet()) {
- //for writers like solr who modify the keys, we want to undo that if we can
- msg.put(cleanField(kv.getKey()), kv.getValue());
+ if(result.failed()) {
+ failed = true;
+ result.getBadResults(buffer);
+ buffer.append(String.format("%d Valid messages processed", docs.size())).append("\n");
+ for (Map<String, Object> doc : docs) {
+ Map<String, Object> msg = new HashMap<>();
+ for (Map.Entry<String, Object> kv : doc.entrySet()) {
+ //for writers like solr who modify the keys, we want to undo that if we can
+ buffer.append(cleanField(kv.getKey())).append(kv.getValue().toString()).append("\n");
+ }
+ }
+ Assert.fail(buffer.toString());
+ }else {
+ for (Map<String, Object> doc : docs) {
+ Map<String, Object> msg = new HashMap<>();
+ for (Map.Entry<String, Object> kv : doc.entrySet()) {
+ //for writers like solr who modify the keys, we want to undo that if we can
+ msg.put(cleanField(kv.getKey()), kv.getValue());
+ }
+ ret.add(msg);
}
- ret.add(msg);
}
return ret;
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ComponentRunner.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ComponentRunner.java b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ComponentRunner.java
index d78fd97..b1f581d 100644
--- a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ComponentRunner.java
+++ b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ComponentRunner.java
@@ -122,7 +122,7 @@ public class ComponentRunner {
}
- public <T> T process(Processor<T> successState) {
+ public <T> ProcessorResult<T> process(Processor<T> successState) {
int retryCount = 0;
long start = System.currentTimeMillis();
while(true) {
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/Processor.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/Processor.java b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/Processor.java
index bbcfb73..55b5299 100644
--- a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/Processor.java
+++ b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/Processor.java
@@ -19,5 +19,5 @@ package org.apache.metron.integration;
public interface Processor<T> {
ReadinessState process(ComponentRunner runner);
- T getResult();
+ ProcessorResult<T> getResult();
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/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
new file mode 100644
index 0000000..3eb4e8f
--- /dev/null
+++ b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/ProcessorResult.java
@@ -0,0 +1,93 @@
+/**
+ * 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.integration;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ProcessorResult<T> {
+ public static class Builder<T>{
+ T result;
+ List<byte[]> processErrors;
+ List<byte[]> processInvalids;
+
+ public Builder(){}
+
+ public Builder withResult(T result){
+ this.result = result;
+ return this;
+ }
+
+ public Builder withProcessErrors(List<byte[]> processErrors){
+ this.processErrors = processErrors;
+ return this;
+ }
+
+ public Builder withProcessInvalids(List<byte[]> processInvalids){
+ this.processInvalids = processInvalids;
+ return this;
+ }
+
+ public ProcessorResult<T> build(){
+ return new ProcessorResult<T>(result,processErrors,processInvalids);
+ }
+
+ }
+
+ T result;
+ List<byte[]> processErrors;
+ List<byte[]> processInvalids;
+ @SuppressWarnings("unchecked")
+ public ProcessorResult(T result,List<byte[]> processErrors, List<byte[]> processInvalids){
+ this.result = result;
+ this.processErrors = processErrors == null ? new ArrayList() : processErrors;
+ this.processInvalids = processInvalids == null ? new ArrayList() : processInvalids;
+ }
+
+ public T getResult(){
+ return result;
+ }
+
+ public List<byte[]> getProcessErrors(){
+ return processErrors;
+ }
+
+ public List<byte[]> getProcessInvalids(){
+ return processInvalids;
+ }
+
+ public boolean failed(){
+ return processErrors.size() > 0 || processInvalids.size() > 0;
+ }
+
+ public void getBadResults(StringBuffer buffer){
+ if(buffer == null){
+ return;
+ }
+ buffer.append(String.format("%d Errors", processErrors.size()));
+ for (byte[] outputMessage : processErrors) {
+ 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/d6d3d457/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 b3d09d0..f4f6529 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
@@ -21,10 +21,7 @@ import junit.framework.Assert;
import org.apache.metron.TestConstants;
import org.apache.metron.common.Constants;
import org.apache.metron.enrichment.integration.components.ConfigUploadComponent;
-import org.apache.metron.integration.BaseIntegrationTest;
-import org.apache.metron.integration.ComponentRunner;
-import org.apache.metron.integration.Processor;
-import org.apache.metron.integration.ReadinessState;
+import org.apache.metron.integration.*;
import org.apache.metron.integration.components.KafkaWithZKComponent;
import org.apache.metron.integration.utils.TestUtils;
import org.apache.metron.parsers.integration.components.ParserTopologyComponent;
@@ -67,45 +64,64 @@ public abstract class ParserIntegrationTest extends BaseIntegrationTest {
.withNumRetries(10)
.build();
runner.start();
- kafkaComponent.writeMessages(sensorType, inputMessages);
- List<byte[]> outputMessages =
- runner.process(new Processor<List<byte[]>>() {
- List<byte[]> messages = null;
+ try {
+ kafkaComponent.writeMessages(sensorType, inputMessages);
+ ProcessorResult<List<byte[]>> result =
+ runner.process(new Processor<List<byte[]>>() {
+ List<byte[]> messages = null;
+ List<byte[]> errors = null;
+ List<byte[]> invalids = null;
- public ReadinessState process(ComponentRunner runner) {
- KafkaWithZKComponent kafkaWithZKComponent = runner.getComponent("kafka", KafkaWithZKComponent.class);
- List<byte[]> outputMessages = kafkaWithZKComponent.readMessages(Constants.ENRICHMENT_TOPIC);
- if (outputMessages.size() == inputMessages.size()) {
- messages = outputMessages;
- return ReadinessState.READY;
- } else {
- return ReadinessState.NOT_READY;
+ public ReadinessState process(ComponentRunner runner) {
+ KafkaWithZKComponent kafkaWithZKComponent = runner.getComponent("kafka", KafkaWithZKComponent.class);
+ List<byte[]> outputMessages = kafkaWithZKComponent.readMessages(Constants.ENRICHMENT_TOPIC);
+ if (outputMessages.size() == inputMessages.size()) {
+ messages = outputMessages;
+ return ReadinessState.READY;
+ } else {
+ errors = kafkaWithZKComponent.readMessages(Constants.ERROR_STREAM);
+ invalids = kafkaWithZKComponent.readMessages(Constants.INVALID_STREAM);
+ if(errors.size() > 0 || invalids.size() > 0) {
+ messages = outputMessages;
+ return ReadinessState.READY;
+ }
+ return ReadinessState.NOT_READY;
+ }
}
- }
- public List<byte[]> getResult() {
- return messages;
- }
- });
- List<ParserValidation> validations = getValidations();
- if (validations == null || validations.isEmpty()) {
- System.out.println("No validations configured for sensorType " + sensorType + ". Dumping parsed messages");
- System.out.println();
- dumpParsedMessages(outputMessages);
- System.out.println();
- Assert.fail();
- } else {
- for (ParserValidation validation : validations) {
- System.out.println("Running " + validation.getName() + " on sensorType " + sensorType);
- validation.validate(sensorType, outputMessages);
+ public ProcessorResult<List<byte[]>> getResult() {
+ ProcessorResult.Builder<List<byte[]>> builder = new ProcessorResult.Builder();
+ return builder.withResult(messages).withProcessErrors(errors).withProcessInvalids(invalids).build();
+ }
+ });
+ List<byte[]> outputMessages = result.getResult();
+ StringBuffer buffer = new StringBuffer();
+ if (result.failed()){
+ result.getBadResults(buffer);
+ buffer.append(String.format("%d Valid Messages Processed", outputMessages.size())).append("\n");
+ dumpParsedMessages(outputMessages,buffer);
+ Assert.fail(buffer.toString());
+ } else {
+ List<ParserValidation> validations = getValidations();
+ if (validations == null || validations.isEmpty()) {
+ buffer.append("No validations configured for sensorType " + sensorType + ". Dumping parsed messages").append("\n");
+ dumpParsedMessages(outputMessages,buffer);
+ Assert.fail(buffer.toString());
+ } else {
+ for (ParserValidation validation : validations) {
+ System.out.println("Running " + validation.getName() + " on sensorType " + sensorType);
+ validation.validate(sensorType, outputMessages);
+ }
+ }
}
+ } finally {
+ runner.stop();
}
- runner.stop();
}
- public void dumpParsedMessages(List<byte[]> outputMessages) {
+ public void dumpParsedMessages(List<byte[]> outputMessages, StringBuffer buffer) {
for (byte[] outputMessage : outputMessages) {
- System.out.println(new String(outputMessage));
+ buffer.append(new String(outputMessage)).append("\n");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java
index 2f359ef..5c810d3 100644
--- a/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java
+++ b/metron-platform/metron-parsers/src/test/java/org/apache/metron/writers/integration/SimpleHbaseEnrichmentWriterIntegrationTest.java
@@ -103,7 +103,7 @@ public class SimpleHbaseEnrichmentWriterIntegrationTest extends BaseIntegrationT
try {
runner.start();
kafkaComponent.writeMessages(sensorType, inputMessages);
- List<LookupKV<EnrichmentKey, EnrichmentValue>> outputMessages =
+ ProcessorResult<List<LookupKV<EnrichmentKey, EnrichmentValue>>> result =
runner.process(new Processor<List<LookupKV<EnrichmentKey, EnrichmentValue>>>() {
List<LookupKV<EnrichmentKey, EnrichmentValue>> messages = null;
@@ -123,8 +123,9 @@ public class SimpleHbaseEnrichmentWriterIntegrationTest extends BaseIntegrationT
return ReadinessState.NOT_READY;
}
- public List<LookupKV<EnrichmentKey, EnrichmentValue>> getResult() {
- return messages;
+ public ProcessorResult<List<LookupKV<EnrichmentKey, EnrichmentValue>>> getResult() {
+ ProcessorResult.Builder<List<LookupKV<EnrichmentKey,EnrichmentValue>>> builder = new ProcessorResult.Builder();
+ return builder.withResult(messages).build();
}
});
Set<String> validIndicators = new HashSet<>(ImmutableList.of("col12", "col22", "col32"));
@@ -142,7 +143,7 @@ public class SimpleHbaseEnrichmentWriterIntegrationTest extends BaseIntegrationT
put("col3", "col33");
}});
}};
- for (LookupKV<EnrichmentKey, EnrichmentValue> kv : outputMessages) {
+ for (LookupKV<EnrichmentKey, EnrichmentValue> kv : result.getResult()) {
Assert.assertTrue(validIndicators.contains(kv.getKey().indicator));
Assert.assertEquals(kv.getValue().getMetadata().get("source.type"), "dummy");
Assert.assertNotNull(kv.getValue().getMetadata().get("timestamp"));
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/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 b9377f3..1fd1775 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
@@ -119,7 +119,7 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
try {
runner.start();
kafkaComponent.writeMessages(sensorType, inputMessages);
- Map<String, List<JSONObject>> outputMessages =
+ ProcessorResult<Map<String, List<JSONObject>>> result =
runner.process(new Processor<Map<String, List<JSONObject>>>() {
Map<String, List<JSONObject>> messages = null;
@@ -139,10 +139,12 @@ public class WriterBoltIntegrationTest extends BaseIntegrationTest {
return ReadinessState.NOT_READY;
}
- public Map<String, List<JSONObject>> getResult() {
- return messages;
+ public ProcessorResult<Map<String, List<JSONObject>>> getResult() {
+ ProcessorResult.Builder<Map<String,List<JSONObject>>> builder = new ProcessorResult.Builder();
+ return builder.withResult(messages).build();
}
});
+ Map<String,List<JSONObject>> outputMessages = result.getResult();
Assert.assertEquals(3, outputMessages.size());
Assert.assertEquals(1, outputMessages.get(Constants.ENRICHMENT_TOPIC).size());
Assert.assertEquals("valid", outputMessages.get(Constants.ENRICHMENT_TOPIC).get(0).get("action"));
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/integration/PcapTopologyIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/integration/PcapTopologyIntegrationTest.java b/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/integration/PcapTopologyIntegrationTest.java
index fcdb108..382a638 100644
--- a/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/integration/PcapTopologyIntegrationTest.java
+++ b/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/integration/PcapTopologyIntegrationTest.java
@@ -35,6 +35,7 @@ import org.apache.kafka.clients.producer.Producer;
import org.apache.metron.common.Constants;
import org.apache.metron.integration.ComponentRunner;
import org.apache.metron.integration.Processor;
+import org.apache.metron.integration.ProcessorResult;
import org.apache.metron.integration.ReadinessState;
import org.apache.metron.integration.components.FluxTopologyComponent;
import org.apache.metron.integration.components.KafkaWithZKComponent;
@@ -264,7 +265,7 @@ public class PcapTopologyIntegrationTest {
}
@Override
- public Void getResult() {
+ public ProcessorResult<Void> getResult() {
return null;
}
});
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/d6d3d457/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 dd9c559..dd38893 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,6 +18,7 @@
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.interfaces.FieldNameConverter;
import org.apache.metron.enrichment.integration.utils.SampleUtil;
@@ -25,6 +26,7 @@ import org.apache.metron.indexing.integration.IndexingIntegrationTest;
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.KafkaWithZKComponent;
import org.apache.metron.solr.integration.components.SolrComponent;
@@ -75,8 +77,10 @@ public class SolrIndexingIntegrationTest extends IndexingIntegrationTest {
public Processor<List<Map<String, Object>>> getProcessor(final List<byte[]> inputMessages) {
return new Processor<List<Map<String, Object>>>() {
List<Map<String, Object>> docs = null;
+ List<byte[]> errors = null;
public ReadinessState process(ComponentRunner runner) {
SolrComponent solrComponent = runner.getComponent("search", SolrComponent.class);
+ KafkaWithZKComponent kafkaWithZKComponent = runner.getComponent("kafka", KafkaWithZKComponent.class);
if (solrComponent.hasCollection(collection)) {
List<Map<String, Object>> docsFromDisk;
try {
@@ -87,6 +91,10 @@ public class SolrIndexingIntegrationTest extends IndexingIntegrationTest {
throw new IllegalStateException("Unable to retrieve indexed documents.", e);
}
if (docs.size() < inputMessages.size() || docs.size() != docsFromDisk.size()) {
+ errors = kafkaWithZKComponent.readMessages(Constants.INDEXING_ERROR_TOPIC);
+ if(errors.size() > 0){
+ return ReadinessState.READY;
+ }
return ReadinessState.NOT_READY;
} else {
return ReadinessState.READY;
@@ -96,8 +104,9 @@ public class SolrIndexingIntegrationTest extends IndexingIntegrationTest {
}
}
- public List<Map<String, Object>> getResult() {
- return docs;
+ public ProcessorResult<List<Map<String, Object>>> getResult() {
+ ProcessorResult.Builder<List<Map<String,Object>>> builder = new ProcessorResult.Builder();
+ return builder.withResult(docs).withProcessErrors(errors).build();
}
};
}