This is an automated email from the ASF dual-hosted git repository. rmerriman pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/metron.git
The following commit(s) were added to refs/heads/master by this push: new 2f2c802 METRON-2026 Remove Storm dependency from metron-common (merrimanr) closes apache/metron#1351 2f2c802 is described below commit 2f2c80244b34e8182f72e6f4dc98a2940194eaa0 Author: merrimanr <merrim...@gmail.com> AuthorDate: Mon Apr 15 16:40:51 2019 -0500 METRON-2026 Remove Storm dependency from metron-common (merrimanr) closes apache/metron#1351 --- .../metron/profiler/storm/ProfileSplitterBolt.java | 2 +- .../metron-common-storm/pom.xml | 89 ++++++++++++++++++++++ .../metron/storm}/common/bolt/ConfiguredBolt.java | 2 +- .../common/bolt/ConfiguredEnrichmentBolt.java | 2 +- .../storm}/common/bolt/ConfiguredIndexingBolt.java | 2 +- .../storm}/common/bolt/ConfiguredParserBolt.java | 2 +- .../storm}/common/bolt/ConfiguredProfilerBolt.java | 2 +- .../storm}/common/message/BytesFromPosition.java | 2 +- .../storm}/common/message/JSONFromField.java | 2 +- .../common/message/JSONFromFieldByReference.java | 2 +- .../storm}/common/message/JSONFromPosition.java | 2 +- .../storm}/common/message/MessageGetStrategy.java | 2 +- .../storm}/common/message/MessageGetters.java | 2 +- .../storm}/common/message/ObjectFromField.java | 2 +- .../common/message/metadata/RawMessageUtil.java} | 65 ++++++---------- .../storm/common/utils/StormErrorUtils.java} | 32 +++++--- .../storm}/common/bolt/BaseConfiguredBoltTest.java | 2 +- .../common/bolt/ConfiguredEnrichmentBoltTest.java | 16 ++-- .../common/bolt/ConfiguredParserBoltTest.java | 13 ++-- .../storm}/common/message/MessageGettersTest.java | 2 +- .../message/metadata/RawMessageUtilTest.java | 9 ++- .../metron/storm}/common/utils/ErrorUtilsTest.java | 5 +- metron-platform/metron-common-streaming/pom.xml | 33 ++++++++ metron-platform/metron-common/pom.xml | 75 ++---------------- .../apache/metron/common/error/MetronError.java | 4 +- .../common/message/metadata/MetadataUtil.java | 53 ------------- .../common/message/metadata/RawMessageUtil.java | 59 -------------- .../org/apache/metron/common/utils/ErrorUtils.java | 20 ----- .../metron/common/writer/BulkMessageWriter.java | 3 +- .../elasticsearch/writer/ElasticsearchWriter.java | 3 +- .../writer/ElasticsearchWriterTest.java | 18 ++--- .../metron/enrichment/bolt/EnrichmentJoinBolt.java | 2 +- .../enrichment/bolt/GenericEnrichmentBolt.java | 10 +-- .../apache/metron/enrichment/bolt/JoinBolt.java | 10 +-- .../apache/metron/enrichment/bolt/SplitBolt.java | 2 +- .../enrichment/bolt/ThreatIntelJoinBolt.java | 2 +- .../enrichment/bolt/UnifiedEnrichmentBolt.java | 12 +-- .../enrichment/bolt/EnrichmentJoinBoltTest.java | 2 +- .../metron/enrichment/bolt/JoinBoltTest.java | 2 +- .../enrichment/bolt/ThreatIntelJoinBoltTest.java | 2 +- .../metron-parsing/metron-parsing-storm/pom.xml | 2 +- .../org/apache/metron/parsers/bolt/ParserBolt.java | 14 ++-- .../org/apache/metron/parsers/bolt/WriterBolt.java | 8 +- .../apache/metron/parsers/bolt/WriterHandler.java | 4 +- .../apache/metron/parsers/bolt/ParserBoltTest.java | 2 +- .../apache/metron/parsers/bolt/WriterBoltTest.java | 8 +- .../integration/validation/StormParserDriver.java | 2 +- .../org/apache/metron/solr/writer/SolrWriter.java | 3 +- .../schema/SchemaValidationIntegrationTest.java | 2 +- .../apache/metron/solr/writer/SolrWriterTest.java | 4 +- metron-platform/metron-writer/pom.xml | 2 +- .../org/apache/metron/writer/AckTuplesPolicy.java | 2 +- .../java/org/apache/metron/writer/NoopWriter.java | 3 +- .../apache/metron/writer/WriterToBulkWriter.java | 3 +- .../metron/writer/bolt/BulkMessageWriterBolt.java | 16 ++-- .../writer/hbase/SimpleHbaseEnrichmentWriter.java | 3 +- .../org/apache/metron/writer/hdfs/HdfsWriter.java | 7 +- .../apache/metron/writer/kafka/KafkaWriter.java | 3 +- .../apache/metron/writer/AckTuplesPolicyTest.java | 2 +- .../metron/writer/BulkWriterComponentTest.java | 1 - .../writer/bolt/BulkMessageWriterBoltTest.java | 12 ++- .../apache/metron/writer/hdfs/HdfsWriterTest.java | 51 ++++++++----- metron-platform/pom.xml | 1 + 63 files changed, 331 insertions(+), 395 deletions(-) diff --git a/metron-analytics/metron-profiler-storm/src/main/java/org/apache/metron/profiler/storm/ProfileSplitterBolt.java b/metron-analytics/metron-profiler-storm/src/main/java/org/apache/metron/profiler/storm/ProfileSplitterBolt.java index ef58ad9..2161910 100644 --- a/metron-analytics/metron-profiler-storm/src/main/java/org/apache/metron/profiler/storm/ProfileSplitterBolt.java +++ b/metron-analytics/metron-profiler-storm/src/main/java/org/apache/metron/profiler/storm/ProfileSplitterBolt.java @@ -20,7 +20,7 @@ package org.apache.metron.profiler.storm; -import org.apache.metron.common.bolt.ConfiguredProfilerBolt; +import org.apache.metron.storm.common.bolt.ConfiguredProfilerBolt; import org.apache.metron.common.configuration.profiler.ProfilerConfig; import org.apache.metron.profiler.DefaultMessageRouter; import org.apache.metron.profiler.MessageRoute; diff --git a/metron-platform/metron-common-streaming/metron-common-storm/pom.xml b/metron-platform/metron-common-streaming/metron-common-storm/pom.xml new file mode 100644 index 0000000..a063163 --- /dev/null +++ b/metron-platform/metron-common-streaming/metron-common-storm/pom.xml @@ -0,0 +1,89 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. + --> + +<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.metron</groupId> + <artifactId>metron-common-streaming</artifactId> + <version>0.7.1</version> + </parent> + <artifactId>metron-common-storm</artifactId> + <name>metron-common-storm</name> + <description>Components common to Storm</description> + <url>https://metron.apache.org/</url> + <properties> + <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> + <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> + <commons.config.version>1.10</commons.config.version> + </properties> + <dependencies> + <dependency> + <groupId>org.apache.metron</groupId> + <artifactId>metron-common</artifactId> + <version>${project.parent.version}</version> + </dependency> + <dependency> + <groupId>org.apache.storm</groupId> + <artifactId>storm-core</artifactId> + <version>${global_storm_version}</version> + <scope>provided</scope> + <exclusions> + <exclusion> + <groupId>org.apache.logging.log4j</groupId> + <artifactId>log4j-core</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.logging.log4j</groupId> + <artifactId>log4j-api</artifactId> + </exclusion> + <exclusion> + <artifactId>servlet-api</artifactId> + <groupId>javax.servlet</groupId> + </exclusion> + <exclusion> + <artifactId>log4j-over-slf4j</artifactId> + <groupId>org.slf4j</groupId> + </exclusion> + <exclusion> + <artifactId>log4j-slf4j-impl</artifactId> + <groupId>org.apache.logging.log4j</groupId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.storm</groupId> + <artifactId>flux-core</artifactId> + <version>${global_flux_version}</version> + </dependency> + <dependency> + <groupId>org.apache.curator</groupId> + <artifactId>curator-client</artifactId> + <version>${global_curator_version}</version> + </dependency> + <dependency> + <groupId>org.apache.curator</groupId> + <artifactId>curator-test</artifactId> + <version>${global_curator_version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.metron</groupId> + <artifactId>metron-test-utilities</artifactId> + <version>${project.parent.version}</version> + <scope>test</scope> + </dependency> + </dependencies> +</project> diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredBolt.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredBolt.java similarity index 99% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredBolt.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredBolt.java index 221edad..a437574 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredBolt.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredBolt.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.bolt; +package org.apache.metron.storm.common.bolt; import java.lang.invoke.MethodHandles; import java.util.Map; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBolt.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredEnrichmentBolt.java similarity index 97% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBolt.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredEnrichmentBolt.java index 2e03a36..bde85bb 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBolt.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredEnrichmentBolt.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.bolt; +package org.apache.metron.storm.common.bolt; import java.io.IOException; import java.lang.invoke.MethodHandles; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredIndexingBolt.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredIndexingBolt.java similarity index 96% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredIndexingBolt.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredIndexingBolt.java index 27e081e..2d650f0 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredIndexingBolt.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredIndexingBolt.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.bolt; +package org.apache.metron.storm.common.bolt; import java.lang.invoke.MethodHandles; import org.apache.metron.common.configuration.IndexingConfigurations; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredParserBolt.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredParserBolt.java similarity index 97% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredParserBolt.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredParserBolt.java index 17b614b..5074271 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredParserBolt.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredParserBolt.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.bolt; +package org.apache.metron.storm.common.bolt; import java.io.IOException; import java.lang.invoke.MethodHandles; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredProfilerBolt.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredProfilerBolt.java similarity index 97% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredProfilerBolt.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredProfilerBolt.java index e4d9f7b..b4207c2 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/bolt/ConfiguredProfilerBolt.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/bolt/ConfiguredProfilerBolt.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.bolt; +package org.apache.metron.storm.common.bolt; import java.lang.invoke.MethodHandles; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/BytesFromPosition.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/BytesFromPosition.java similarity index 96% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/message/BytesFromPosition.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/BytesFromPosition.java index 56c6490..b47f43c 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/BytesFromPosition.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/BytesFromPosition.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message; +package org.apache.metron.storm.common.message; import org.apache.storm.tuple.Tuple; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/JSONFromField.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/JSONFromField.java similarity index 96% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/message/JSONFromField.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/JSONFromField.java index 39fe9dd..b8df105 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/JSONFromField.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/JSONFromField.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message; +package org.apache.metron.storm.common.message; import org.apache.storm.tuple.Tuple; import org.json.simple.JSONObject; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/JSONFromFieldByReference.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/JSONFromFieldByReference.java similarity index 96% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/message/JSONFromFieldByReference.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/JSONFromFieldByReference.java index a0d4b7d..1868bc7 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/JSONFromFieldByReference.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/JSONFromFieldByReference.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message; +package org.apache.metron.storm.common.message; import org.apache.storm.tuple.Tuple; import org.json.simple.JSONObject; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/JSONFromPosition.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/JSONFromPosition.java similarity index 97% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/message/JSONFromPosition.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/JSONFromPosition.java index 15f0447..b81373e 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/JSONFromPosition.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/JSONFromPosition.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message; +package org.apache.metron.storm.common.message; import org.apache.commons.io.Charsets; import org.apache.storm.tuple.Tuple; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/MessageGetStrategy.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/MessageGetStrategy.java similarity index 95% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/message/MessageGetStrategy.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/MessageGetStrategy.java index 0595ce1..dfe8fff 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/MessageGetStrategy.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/MessageGetStrategy.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message; +package org.apache.metron.storm.common.message; import org.apache.storm.tuple.Tuple; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/MessageGetters.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/MessageGetters.java similarity index 98% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/message/MessageGetters.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/MessageGetters.java index 46bb406..36ad1c9 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/MessageGetters.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/MessageGetters.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message; +package org.apache.metron.storm.common.message; import org.apache.metron.stellar.common.utils.ConversionUtils; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/ObjectFromField.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/ObjectFromField.java similarity index 96% copy from metron-platform/metron-common/src/main/java/org/apache/metron/common/message/ObjectFromField.java copy to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/ObjectFromField.java index 120c09c..dced7d6 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/ObjectFromField.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/ObjectFromField.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message; +package org.apache.metron.storm.common.message; import org.apache.storm.tuple.Tuple; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/MetadataUtil.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/metadata/RawMessageUtil.java similarity index 67% copy from metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/MetadataUtil.java copy to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/metadata/RawMessageUtil.java index 3034ddd..afd3a87 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/MetadataUtil.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/message/metadata/RawMessageUtil.java @@ -15,9 +15,12 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message.metadata; +package org.apache.metron.storm.common.message.metadata; import org.apache.commons.lang3.StringUtils; +import org.apache.metron.common.message.metadata.MetadataUtil; +import org.apache.metron.common.message.metadata.RawMessage; +import org.apache.metron.common.message.metadata.RawMessageStrategy; import org.apache.metron.common.utils.JSONUtils; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Tuple; @@ -29,51 +32,29 @@ import java.lang.invoke.MethodHandles; import java.util.HashMap; import java.util.Map; -/** - * Captures some common utility methods around metadata manipulation. - */ -public enum MetadataUtil { +public enum RawMessageUtil { + INSTANCE; private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); - /** - * The default metadata prefix. - */ - public static final String METADATA_PREFIX = "metron.metadata"; - /** - * The config key for defining the prefix. - */ - public static final String METADATA_PREFIX_CONFIG = "metadataPrefix"; - static final int KEY_INDEX = 1; - /** - * Return the prefix that we want to use for metadata keys. This comes from the config and is defaulted to - * 'metron.metadata'. - * - * @param config The rawMessageStrategyConfig - * @return the prefix for metadata keys - */ - public String getMetadataPrefix(Map<String, Object> config) { - String prefix = (String) config.getOrDefault(METADATA_PREFIX_CONFIG, METADATA_PREFIX); - if(StringUtils.isEmpty(prefix)) { - return null; - } - return prefix; - } + static final int KEY_INDEX = 1; /** - * Take a field and prefix it with the metadata key. - * - * @param prefix The metadata prefix to use (e.g. 'foo') - * @param key The key name (e.g. my_field) - * @return The prefixed key separated by a . (e.g. foo.my_field) + * Extract the raw message given the strategy, the tuple and the metadata configs. + * @param strategy The {@link RawMessageStrategy} to use for extraction + * @param t The tuple to pull the message from + * @param rawMessage The raw message in bytes + * @param readMetadata True if read metadata, false otherwise + * @param config The config to use during extraction + * @return The resulting {@link RawMessage} */ - public String prefixKey(String prefix, String key) { - if(StringUtils.isEmpty(prefix)) { - return key; - } - else { - return prefix + "." + key; + public RawMessage getRawMessage(RawMessageStrategy strategy, Tuple t, byte[] rawMessage, boolean readMetadata, Map<String, Object> config) { + Map<String, Object> metadata = new HashMap<>(); + if(readMetadata) { + String prefix = MetadataUtil.INSTANCE.getMetadataPrefix(config); + metadata = extractMetadata(prefix, t); } + return strategy.get(metadata, rawMessage, readMetadata, config); } /** @@ -101,7 +82,7 @@ public enum MetadataUtil { String envMetadataFieldName = tupleFields.get(i); Object envMetadataFieldValue = t.getValue(i); if (!StringUtils.isEmpty(envMetadataFieldName) && envMetadataFieldValue != null) { - metadata.put(prefixKey(prefix, envMetadataFieldName), envMetadataFieldValue); + metadata.put(MetadataUtil.INSTANCE.prefixKey(prefix, envMetadataFieldName), envMetadataFieldValue); } } byte[] keyObj = t.getBinary(KEY_INDEX); @@ -111,7 +92,7 @@ public enum MetadataUtil { if (!StringUtils.isEmpty(keyStr)) { Map<String, Object> rawMetadata = JSONUtils.INSTANCE.load(keyStr, JSONUtils.MAP_SUPPLIER); for (Map.Entry<String, Object> kv : rawMetadata.entrySet()) { - metadata.put(prefixKey(prefix, kv.getKey()), kv.getValue()); + metadata.put(MetadataUtil.INSTANCE.prefixKey(prefix, kv.getKey()), kv.getValue()); } } @@ -122,4 +103,6 @@ public enum MetadataUtil { } return metadata; } + + } diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/ObjectFromField.java b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/utils/StormErrorUtils.java similarity index 51% rename from metron-platform/metron-common/src/main/java/org/apache/metron/common/message/ObjectFromField.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/utils/StormErrorUtils.java index 120c09c..cd7f93d 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/ObjectFromField.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/main/java/org/apache/metron/storm/common/utils/StormErrorUtils.java @@ -15,22 +15,30 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message; -import org.apache.storm.tuple.Tuple; +package org.apache.metron.storm.common.utils; -public class ObjectFromField implements MessageGetStrategy { +import org.apache.metron.common.Constants; +import org.apache.metron.common.error.MetronError; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.tuple.Values; - private String fieldValue = "message"; +import java.util.Optional; - public ObjectFromField() {}; +public class StormErrorUtils { + /** + * Handles a {@link MetronError} that occurs. + * + * @param collector The Storm output collector being reported to + * @param error The error that occurred + */ + public static void handleError(OutputCollector collector, MetronError error) + { + collector.emit(Constants.ERROR_STREAM, new Values(error.getJSONObject())); + Optional<Throwable> throwable = error.getThrowable(); + if (throwable.isPresent()) { + collector.reportError(throwable.get()); + } - public ObjectFromField(String fieldValue) { - this.fieldValue = fieldValue; - } - - @Override - public Object get(Tuple tuple) { - return tuple.getValueByField(fieldValue); } } diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/BaseConfiguredBoltTest.java b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/bolt/BaseConfiguredBoltTest.java similarity index 97% rename from metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/BaseConfiguredBoltTest.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/bolt/BaseConfiguredBoltTest.java index f9901cd..ac2d1e8 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/BaseConfiguredBoltTest.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/bolt/BaseConfiguredBoltTest.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.bolt; +package org.apache.metron.storm.common.bolt; import org.apache.metron.test.bolt.BaseBoltTest; import org.junit.Assert; diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBoltTest.java b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/bolt/ConfiguredEnrichmentBoltTest.java similarity index 91% rename from metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBoltTest.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/bolt/ConfiguredEnrichmentBoltTest.java index 44612cd..256415a 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredEnrichmentBoltTest.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/bolt/ConfiguredEnrichmentBoltTest.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.bolt; +package org.apache.metron.storm.common.bolt; import org.apache.log4j.Level; import org.apache.metron.test.utils.UnitTestHelper; @@ -37,6 +37,10 @@ import java.util.Set; public class ConfiguredEnrichmentBoltTest extends BaseConfiguredBoltTest { + private static final String sampleConfigPath = "../" + TestConstants.SAMPLE_CONFIG_PATH; + private static final String enrichmentsConfigPath = "../" + TestConstants.ENRICHMENTS_CONFIGS_PATH; + private static final String parserConfigsPath = "../" + TestConstants.PARSER_CONFIGS_PATH; + private Set<String> enrichmentConfigurationTypes = new HashSet<>(); private String zookeeperUrl; @@ -64,15 +68,15 @@ public class ConfiguredEnrichmentBoltTest extends BaseConfiguredBoltTest { public void setupConfiguration() throws Exception { TestingServer testZkServer = new TestingServer(true); this.zookeeperUrl = testZkServer.getConnectString(); - byte[] globalConfig = ConfigurationsUtils.readGlobalConfigFromFile(TestConstants.SAMPLE_CONFIG_PATH); + byte[] globalConfig = ConfigurationsUtils.readGlobalConfigFromFile(sampleConfigPath); ConfigurationsUtils.writeGlobalConfigToZookeeper(globalConfig, zookeeperUrl); enrichmentConfigurationTypes.add(ConfigurationType.GLOBAL.getTypeName()); - Map<String, byte[]> sensorEnrichmentConfigs = ConfigurationsUtils.readSensorEnrichmentConfigsFromFile(TestConstants.ENRICHMENTS_CONFIGS_PATH); + Map<String, byte[]> sensorEnrichmentConfigs = ConfigurationsUtils.readSensorEnrichmentConfigsFromFile(enrichmentsConfigPath); for (String sensorType : sensorEnrichmentConfigs.keySet()) { ConfigurationsUtils.writeSensorEnrichmentConfigToZookeeper(sensorType, sensorEnrichmentConfigs.get(sensorType), zookeeperUrl); enrichmentConfigurationTypes.add(sensorType); } - Map<String, byte[]> sensorParserConfigs = ConfigurationsUtils.readSensorParserConfigsFromFile(TestConstants.PARSER_CONFIGS_PATH); + Map<String, byte[]> sensorParserConfigs = ConfigurationsUtils.readSensorParserConfigsFromFile(parserConfigsPath); for (String sensorType : sensorParserConfigs.keySet()) { ConfigurationsUtils.writeSensorParserConfigToZookeeper(sensorType, sensorParserConfigs.get(sensorType), zookeeperUrl); } @@ -90,8 +94,8 @@ public class ConfiguredEnrichmentBoltTest extends BaseConfiguredBoltTest { UnitTestHelper.setLog4jLevel(ConfiguredBolt.class, Level.ERROR); configsUpdated = new HashSet<>(); - sampleConfigurations.updateGlobalConfig(ConfigurationsUtils.readGlobalConfigFromFile(TestConstants.SAMPLE_CONFIG_PATH)); - Map<String, byte[]> sensorEnrichmentConfigs = ConfigurationsUtils.readSensorEnrichmentConfigsFromFile(TestConstants.ENRICHMENTS_CONFIGS_PATH); + sampleConfigurations.updateGlobalConfig(ConfigurationsUtils.readGlobalConfigFromFile(sampleConfigPath)); + Map<String, byte[]> sensorEnrichmentConfigs = ConfigurationsUtils.readSensorEnrichmentConfigsFromFile(enrichmentsConfigPath); for (String sensorType : sensorEnrichmentConfigs.keySet()) { sampleConfigurations.updateSensorEnrichmentConfig(sensorType, sensorEnrichmentConfigs.get(sensorType)); } diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredParserBoltTest.java b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/bolt/ConfiguredParserBoltTest.java similarity index 94% rename from metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredParserBoltTest.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/bolt/ConfiguredParserBoltTest.java index 3deba78..66668d1 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/bolt/ConfiguredParserBoltTest.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/bolt/ConfiguredParserBoltTest.java @@ -15,11 +15,10 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.bolt; +package org.apache.metron.storm.common.bolt; import org.apache.log4j.Level; import org.apache.metron.common.configuration.FieldValidator; -import org.apache.metron.common.field.validation.FieldValidation; import org.apache.metron.test.utils.UnitTestHelper; import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.tuple.Tuple; @@ -67,14 +66,14 @@ public class ConfiguredParserBoltTest extends BaseConfiguredBoltTest { public void setupConfiguration() throws Exception { TestingServer testZkServer = new TestingServer(true); this.zookeeperUrl = testZkServer.getConnectString(); - byte[] globalConfig = ConfigurationsUtils.readGlobalConfigFromFile(TestConstants.SAMPLE_CONFIG_PATH); + byte[] globalConfig = ConfigurationsUtils.readGlobalConfigFromFile("../" + TestConstants.SAMPLE_CONFIG_PATH); ConfigurationsUtils.writeGlobalConfigToZookeeper(globalConfig, zookeeperUrl); parserConfigurationTypes.add(ConfigurationType.GLOBAL.getTypeName()); - Map<String, byte[]> sensorEnrichmentConfigs = ConfigurationsUtils.readSensorEnrichmentConfigsFromFile(TestConstants.ENRICHMENTS_CONFIGS_PATH); + Map<String, byte[]> sensorEnrichmentConfigs = ConfigurationsUtils.readSensorEnrichmentConfigsFromFile("../" + TestConstants.ENRICHMENTS_CONFIGS_PATH); for (String sensorType : sensorEnrichmentConfigs.keySet()) { ConfigurationsUtils.writeSensorEnrichmentConfigToZookeeper(sensorType, sensorEnrichmentConfigs.get(sensorType), zookeeperUrl); } - Map<String, byte[]> sensorParserConfigs = ConfigurationsUtils.readSensorParserConfigsFromFile(TestConstants.PARSER_CONFIGS_PATH); + Map<String, byte[]> sensorParserConfigs = ConfigurationsUtils.readSensorParserConfigsFromFile("../" + TestConstants.PARSER_CONFIGS_PATH); for (String sensorType : sensorParserConfigs.keySet()) { ConfigurationsUtils.writeSensorParserConfigToZookeeper(sensorType, sensorParserConfigs.get(sensorType), zookeeperUrl); parserConfigurationTypes.add(sensorType); @@ -93,8 +92,8 @@ public class ConfiguredParserBoltTest extends BaseConfiguredBoltTest { UnitTestHelper.setLog4jLevel(ConfiguredBolt.class, Level.ERROR); configsUpdated = new HashSet<>(); - sampleConfigurations.updateGlobalConfig(ConfigurationsUtils.readGlobalConfigFromFile(TestConstants.SAMPLE_CONFIG_PATH)); - Map<String, byte[]> sensorParserConfigs = ConfigurationsUtils.readSensorParserConfigsFromFile(TestConstants.PARSER_CONFIGS_PATH); + sampleConfigurations.updateGlobalConfig(ConfigurationsUtils.readGlobalConfigFromFile("../" + TestConstants.SAMPLE_CONFIG_PATH)); + Map<String, byte[]> sensorParserConfigs = ConfigurationsUtils.readSensorParserConfigsFromFile("../" + TestConstants.PARSER_CONFIGS_PATH); for (String sensorType : sensorParserConfigs.keySet()) { sampleConfigurations.updateSensorParserConfig(sensorType, sensorParserConfigs.get(sensorType)); } diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/message/MessageGettersTest.java b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/message/MessageGettersTest.java similarity index 99% rename from metron-platform/metron-common/src/test/java/org/apache/metron/common/message/MessageGettersTest.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/message/MessageGettersTest.java index ea7583a..aa3c0a0 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/message/MessageGettersTest.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/message/MessageGettersTest.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message; +package org.apache.metron.storm.common.message; import org.apache.storm.tuple.Tuple; import org.json.simple.JSONObject; diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/message/metadata/RawMessageUtilTest.java b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/message/metadata/RawMessageUtilTest.java similarity index 96% rename from metron-platform/metron-common/src/test/java/org/apache/metron/common/message/metadata/RawMessageUtilTest.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/message/metadata/RawMessageUtilTest.java index e5fd80f..cf51f02 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/message/metadata/RawMessageUtilTest.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/message/metadata/RawMessageUtilTest.java @@ -15,18 +15,21 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.message.metadata; +package org.apache.metron.storm.common.message.metadata; -import com.google.common.base.Joiner; import com.google.common.collect.ImmutableMap; import org.adrianwalker.multilinestring.Multiline; import org.apache.metron.common.Constants; +import org.apache.metron.common.message.metadata.EnvelopedRawMessageStrategy; +import org.apache.metron.common.message.metadata.RawMessage; +import org.apache.metron.common.message.metadata.RawMessageStrategies; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Tuple; import org.json.simple.JSONObject; import org.junit.Assert; import org.junit.Test; +import org.apache.metron.common.message.metadata.MetadataUtil; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -54,7 +57,7 @@ public class RawMessageUtilTest { } when(t.getFields()).thenReturn(f); - when(t.getBinary(eq(MetadataUtil.KEY_INDEX))).thenReturn(metadata.getBytes()); + when(t.getBinary(eq(RawMessageUtil.KEY_INDEX))).thenReturn(metadata.getBytes()); return t; } diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/ErrorUtilsTest.java b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/utils/ErrorUtilsTest.java similarity index 95% rename from metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/ErrorUtilsTest.java rename to metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/utils/ErrorUtilsTest.java index 77ea9da..037a06c 100644 --- a/metron-platform/metron-common/src/test/java/org/apache/metron/common/utils/ErrorUtilsTest.java +++ b/metron-platform/metron-common-streaming/metron-common-storm/src/test/java/org/apache/metron/storm/common/utils/ErrorUtilsTest.java @@ -15,10 +15,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.metron.common.utils; +package org.apache.metron.storm.common.utils; import org.apache.metron.common.Constants; import org.apache.metron.common.error.MetronError; +import org.apache.metron.common.utils.ErrorUtils; import org.apache.metron.test.error.MetronErrorJSONMatcher; import org.apache.storm.task.OutputCollector; import org.junit.Rule; @@ -79,7 +80,7 @@ public class ErrorUtilsTest { MetronError error = new MetronError().withMessage("error message").withThrowable(e); OutputCollector collector = mock(OutputCollector.class); - ErrorUtils.handleError(collector, error); + StormErrorUtils.handleError(collector, error); verify(collector, times(1)).emit(eq(Constants.ERROR_STREAM), argThat(new MetronErrorJSONMatcher(error.getJSONObject()))); verify(collector, times(1)).reportError(any()); } diff --git a/metron-platform/metron-common-streaming/pom.xml b/metron-platform/metron-common-streaming/pom.xml new file mode 100644 index 0000000..4746b9e --- /dev/null +++ b/metron-platform/metron-common-streaming/pom.xml @@ -0,0 +1,33 @@ +<?xml version="1.0" encoding="UTF-8"?> +<project xmlns="http://maven.apache.org/POM/4.0.0" + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + <modelVersion>4.0.0</modelVersion> + <artifactId>metron-common-streaming</artifactId> + <packaging>pom</packaging> + <name>metron-common-streaming</name> + <parent> + <artifactId>metron-platform</artifactId> + <groupId>org.apache.metron</groupId> + <version>0.7.1</version> + </parent> + <description>Common modules for Streaming platforms</description> + <url>https://metron.apache.org/</url> + <scm> + <connection>scm:git:https://gitbox.apache.org/repos/asf/metron.git</connection> + <developerConnection>scm:git:https://gitbox.apache.org/repos/asf/metron.git</developerConnection> + <tag>HEAD</tag> + <url>https://gitbox.apache.org/repos/asf/metron.git</url> + </scm> + + <licenses> + <license> + <name>The Apache Software License, Version 2.0</name> + <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url> + <distribution>repo</distribution> + </license> + </licenses> + <modules> + <module>metron-common-storm</module> + </modules> +</project> \ No newline at end of file diff --git a/metron-platform/metron-common/pom.xml b/metron-platform/metron-common/pom.xml index 4735044..6835933 100644 --- a/metron-platform/metron-common/pom.xml +++ b/metron-platform/metron-common/pom.xml @@ -29,6 +29,7 @@ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <commons.config.version>1.10</commons.config.version> <guava_version>${global_guava_version}</guava_version> + <asm.version>5.0.3</asm.version> </properties> <repositories> <repository> @@ -85,34 +86,6 @@ <version>${global_antlr_version}</version> </dependency> <dependency> - <groupId>org.apache.storm</groupId> - <artifactId>storm-core</artifactId> - <version>${global_storm_version}</version> - <scope>provided</scope> - <exclusions> - <exclusion> - <groupId>org.apache.logging.log4j</groupId> - <artifactId>log4j-core</artifactId> - </exclusion> - <exclusion> - <groupId>org.apache.logging.log4j</groupId> - <artifactId>log4j-api</artifactId> - </exclusion> - <exclusion> - <artifactId>servlet-api</artifactId> - <groupId>javax.servlet</groupId> - </exclusion> - <exclusion> - <artifactId>log4j-over-slf4j</artifactId> - <groupId>org.slf4j</groupId> - </exclusion> - <exclusion> - <artifactId>log4j-slf4j-impl</artifactId> - <groupId>org.apache.logging.log4j</groupId> - </exclusion> - </exclusions> - </dependency> - <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.10</artifactId> <version>${global_kafka_version}</version> @@ -218,18 +191,6 @@ <scope>provided</scope> </dependency> <dependency> - <groupId>org.apache.hbase</groupId> - <artifactId>hbase-common</artifactId> - <version>${global_hbase_version}</version> - <exclusions> - <exclusion> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-mapreduce-client-core</artifactId> - </exclusion> - </exclusions> - <scope>provided</scope> - </dependency> - <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-auth</artifactId> <version>${global_hadoop_version}</version> @@ -257,33 +218,9 @@ <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> - </exclusions> - </dependency> - <dependency> - <groupId>org.apache.hbase</groupId> - <artifactId>hbase-client</artifactId> - <version>${global_hbase_version}</version> - <scope>provided</scope> - <exclusions> - <exclusion> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-auth</artifactId> - </exclusion> - <exclusion> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-common</artifactId> - </exclusion> - <exclusion> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-log4j12</artifactId> - </exclusion> - <exclusion> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </exclusion> <exclusion> - <groupId>com.google.guava</groupId> - <artifactId>guava</artifactId> + <groupId>asm</groupId> + <artifactId>asm</artifactId> </exclusion> </exclusions> </dependency> @@ -303,9 +240,9 @@ <version>${global_jackson_version}</version> </dependency> <dependency> - <groupId>org.apache.storm</groupId> - <artifactId>flux-core</artifactId> - <version>${global_flux_version}</version> + <groupId>org.ow2.asm</groupId> + <artifactId>asm</artifactId> + <version>${asm.version}</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/error/MetronError.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/error/MetronError.java index df21c3b..46fe3b2 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/error/MetronError.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/error/MetronError.java @@ -23,6 +23,7 @@ import static org.apache.metron.common.Constants.ErrorFields; import java.net.InetAddress; import java.net.UnknownHostException; +import java.nio.charset.Charset; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -34,7 +35,6 @@ import java.util.Set; import java.util.UUID; import org.apache.commons.lang.exception.ExceptionUtils; -import org.apache.hadoop.hbase.util.Bytes; import org.apache.metron.common.Constants; import org.apache.metron.common.Constants.ErrorType; import org.apache.metron.common.utils.HashUtils; @@ -180,7 +180,7 @@ public class MetronError { // It's unclear if we need a rawMessageBytes field so commenting out for now //String rawMessageBytesField = rawMessages.size() == 1 ? ErrorFields.RAW_MESSAGE_BYTES.getName() : ErrorFields.RAW_MESSAGE_BYTES.getName() + "_" + i; if(rawMessage instanceof byte[]) { - errorMessage.put(rawMessageField, Bytes.toString((byte[])rawMessage)); + errorMessage.put(rawMessageField, new String((byte[])rawMessage, Charset.forName("UTF-8"))); //errorMessage.put(rawMessageBytesField, com.google.common.primitives.Bytes.asList((byte[])rawMessage)); } else if (rawMessage instanceof JSONObject) { JSONObject rawMessageJSON = (JSONObject) rawMessage; diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/MetadataUtil.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/MetadataUtil.java index 3034ddd..628ef50 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/MetadataUtil.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/MetadataUtil.java @@ -18,15 +18,10 @@ package org.apache.metron.common.message.metadata; import org.apache.commons.lang3.StringUtils; -import org.apache.metron.common.utils.JSONUtils; -import org.apache.storm.tuple.Fields; -import org.apache.storm.tuple.Tuple; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; import java.lang.invoke.MethodHandles; -import java.util.HashMap; import java.util.Map; /** @@ -43,7 +38,6 @@ public enum MetadataUtil { * The config key for defining the prefix. */ public static final String METADATA_PREFIX_CONFIG = "metadataPrefix"; - static final int KEY_INDEX = 1; /** * Return the prefix that we want to use for metadata keys. This comes from the config and is defaulted to @@ -75,51 +69,4 @@ public enum MetadataUtil { return prefix + "." + key; } } - - /** - * Default extraction of metadata. This handles looking in the normal places for metadata - * <ul> - * <li>The kafka key</li> - * <li>The tuple fields outside of the value (e.g. the topic)</li> - * </ul> - * - * <p>In addition to extracting the metadata into a map, it applies the appropriate prefix (as configured in the rawMessageStrategyConfig). - * @param prefix The prefix of the metadata keys - * @param t The tuple to get metadata from - * @return A map containing the metadata - */ - public Map<String, Object> extractMetadata(String prefix, Tuple t) { - Map<String, Object> metadata = new HashMap<>(); - if(t == null) { - return metadata; - } - Fields tupleFields = t.getFields(); - if(tupleFields == null) { - return metadata; - } - for (int i = 2; i < tupleFields.size(); ++i) { - String envMetadataFieldName = tupleFields.get(i); - Object envMetadataFieldValue = t.getValue(i); - if (!StringUtils.isEmpty(envMetadataFieldName) && envMetadataFieldValue != null) { - metadata.put(prefixKey(prefix, envMetadataFieldName), envMetadataFieldValue); - } - } - byte[] keyObj = t.getBinary(KEY_INDEX); - String keyStr = null; - try { - keyStr = keyObj == null ? null : new String(keyObj); - if (!StringUtils.isEmpty(keyStr)) { - Map<String, Object> rawMetadata = JSONUtils.INSTANCE.load(keyStr, JSONUtils.MAP_SUPPLIER); - for (Map.Entry<String, Object> kv : rawMetadata.entrySet()) { - metadata.put(prefixKey(prefix, kv.getKey()), kv.getValue()); - } - - } - } catch (IOException e) { - String reason = "Unable to parse metadata; expected JSON Map: " + (keyStr == null ? "NON-STRING!" : keyStr); - LOG.error(reason, e); - throw new IllegalStateException(reason, e); - } - return metadata; - } } diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/RawMessageUtil.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/RawMessageUtil.java deleted file mode 100644 index 3bd9915..0000000 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/message/metadata/RawMessageUtil.java +++ /dev/null @@ -1,59 +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.common.message.metadata; - -import com.google.common.base.Joiner; -import org.apache.commons.lang3.StringUtils; -import org.apache.metron.common.Constants; -import org.apache.metron.common.utils.JSONUtils; -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.IOException; -import java.lang.invoke.MethodHandles; -import java.util.HashMap; -import java.util.Map; - -public enum RawMessageUtil { - - INSTANCE; - - - /** - * Extract the raw message given the strategy, the tuple and the metadata configs. - * @param strategy The {@link RawMessageStrategy} to use for extraction - * @param t The tuple to pull the message from - * @param rawMessage The raw message in bytes - * @param readMetadata True if read metadata, false otherwise - * @param config The config to use during extraction - * @return The resulting {@link RawMessage} - */ - public RawMessage getRawMessage(RawMessageStrategy strategy, Tuple t, byte[] rawMessage, boolean readMetadata, Map<String, Object> config) { - Map<String, Object> metadata = new HashMap<>(); - if(readMetadata) { - String prefix = MetadataUtil.INSTANCE.getMetadataPrefix(config); - metadata = MetadataUtil.INSTANCE.extractMetadata(prefix, t); - } - return strategy.get(metadata, rawMessage, readMetadata, config); - } - - -} diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/ErrorUtils.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/ErrorUtils.java index e3ad306..bd666c3 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/ErrorUtils.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/utils/ErrorUtils.java @@ -26,10 +26,6 @@ import java.lang.management.ThreadMXBean; import java.util.Optional; import java.util.function.Function; import org.apache.commons.lang3.tuple.Pair; -import org.apache.metron.common.Constants; -import org.apache.metron.common.error.MetronError; -import org.apache.storm.task.OutputCollector; -import org.apache.storm.tuple.Values; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -90,22 +86,6 @@ public class ErrorUtils { } /** - * Handles a {@link MetronError} that occurs. - * - * @param collector The Storm output collector being reported to - * @param error The error that occurred - */ - public static void handleError(OutputCollector collector, MetronError error) - { - collector.emit(Constants.ERROR_STREAM, new Values(error.getJSONObject())); - Optional<Throwable> throwable = error.getThrowable(); - if (throwable.isPresent()) { - collector.reportError(throwable.get()); - } - - } - - /** * Generates a string version of a thread dump. * * @return String of the thread dump diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/writer/BulkMessageWriter.java b/metron-platform/metron-common/src/main/java/org/apache/metron/common/writer/BulkMessageWriter.java index 7f92d37..51bc599 100644 --- a/metron-platform/metron-common/src/main/java/org/apache/metron/common/writer/BulkMessageWriter.java +++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/writer/BulkMessageWriter.java @@ -17,7 +17,6 @@ */ package org.apache.metron.common.writer; -import org.apache.storm.task.TopologyContext; import org.apache.metron.common.configuration.writer.WriterConfiguration; import java.io.Serializable; @@ -26,7 +25,7 @@ import java.util.Map; public interface BulkMessageWriter<MESSAGE_T> extends AutoCloseable, Serializable { - void init(Map stormConf, TopologyContext topologyContext, WriterConfiguration config) throws Exception; + void init(Map stormConf, WriterConfiguration config) throws Exception; /** * Writes the messages to a particular output (e.g. Elasticsearch). A response is returned with successful and failed message ids. diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriter.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriter.java index 7d14c11..b7814b6 100644 --- a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriter.java +++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriter.java @@ -33,7 +33,6 @@ import org.apache.metron.elasticsearch.client.ElasticsearchClient; import org.apache.metron.elasticsearch.client.ElasticsearchClientFactory; import org.apache.metron.elasticsearch.utils.ElasticsearchUtils; import org.apache.metron.stellar.common.utils.ConversionUtils; -import org.apache.storm.task.TopologyContext; import org.json.simple.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -75,7 +74,7 @@ public class ElasticsearchWriter implements BulkMessageWriter<JSONObject>, Seria private SimpleDateFormat dateFormat; @Override - public void init(Map stormConf, TopologyContext topologyContext, WriterConfiguration configurations) { + public void init(Map stormConf, WriterConfiguration configurations) { Map<String, Object> globalConfiguration = configurations.getGlobalConfig(); dateFormat = ElasticsearchUtils.getIndexFormat(globalConfiguration); diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriterTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriterTest.java index 3d6a3fa..ba5cfe0 100644 --- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriterTest.java +++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriterTest.java @@ -25,7 +25,6 @@ import org.apache.metron.common.writer.BulkWriterResponse; import org.apache.metron.common.writer.MessageId; import org.apache.metron.elasticsearch.bulk.BulkDocumentWriter; import org.apache.metron.elasticsearch.bulk.BulkDocumentWriterResults; -import org.apache.storm.task.TopologyContext; import org.json.simple.JSONObject; import org.junit.Before; import org.junit.Test; @@ -47,13 +46,10 @@ import static org.mockito.Mockito.when; public class ElasticsearchWriterTest { Map stormConf; - TopologyContext topologyContext; WriterConfiguration writerConfiguration; @Before public void setup() { - topologyContext = mock(TopologyContext.class); - writerConfiguration = mock(WriterConfiguration.class); when(writerConfiguration.getGlobalConfig()).thenReturn(globals()); @@ -74,7 +70,7 @@ public class ElasticsearchWriterTest { // attempt to write ElasticsearchWriter esWriter = new ElasticsearchWriter(); esWriter.setDocumentWriter(docWriter); - esWriter.init(stormConf, topologyContext, writerConfiguration); + esWriter.init(stormConf, writerConfiguration); BulkWriterResponse response = esWriter.write("bro", writerConfiguration, messages); // response should only contain successes @@ -98,7 +94,7 @@ public class ElasticsearchWriterTest { // attempt to write ElasticsearchWriter esWriter = new ElasticsearchWriter(); esWriter.setDocumentWriter(docWriter); - esWriter.init(stormConf, topologyContext, writerConfiguration); + esWriter.init(stormConf, writerConfiguration); BulkWriterResponse response = esWriter.write("bro", writerConfiguration, messages); // response should only contain successes @@ -123,7 +119,7 @@ public class ElasticsearchWriterTest { // attempt to write ElasticsearchWriter esWriter = new ElasticsearchWriter(); esWriter.setDocumentWriter(docWriter); - esWriter.init(stormConf, topologyContext, writerConfiguration); + esWriter.init(stormConf, writerConfiguration); BulkWriterResponse response = esWriter.write("bro", writerConfiguration, messages); // the writer response should only contain failures @@ -151,7 +147,7 @@ public class ElasticsearchWriterTest { // attempt to write ElasticsearchWriter esWriter = new ElasticsearchWriter(); esWriter.setDocumentWriter(docWriter); - esWriter.init(stormConf, topologyContext, writerConfiguration); + esWriter.init(stormConf, writerConfiguration); BulkWriterResponse response = esWriter.write("bro", writerConfiguration, messages); // the writer response should only contain failures @@ -180,7 +176,7 @@ public class ElasticsearchWriterTest { // attempt to write ElasticsearchWriter esWriter = new ElasticsearchWriter(); esWriter.setDocumentWriter(docWriter); - esWriter.init(stormConf, topologyContext, writerConfiguration); + esWriter.init(stormConf, writerConfiguration); BulkWriterResponse response = esWriter.write("bro", writerConfiguration, messages); // response should contain some successes and some failures @@ -214,7 +210,7 @@ public class ElasticsearchWriterTest { // attempt to write ElasticsearchWriter esWriter = new ElasticsearchWriter(); esWriter.setDocumentWriter(docWriter); - esWriter.init(stormConf, topologyContext, writerConfiguration); + esWriter.init(stormConf, writerConfiguration); BulkWriterResponse response = esWriter.write("bro", writerConfiguration, messages); // response should only contain successes @@ -239,7 +235,7 @@ public class ElasticsearchWriterTest { // attempt to write ElasticsearchWriter esWriter = new ElasticsearchWriter(); esWriter.setDocumentWriter(docWriter); - esWriter.init(stormConf, topologyContext, writerConfiguration); + esWriter.init(stormConf, writerConfiguration); BulkWriterResponse response = esWriter.write("bro", writerConfiguration, messages); // response should only contain successes diff --git a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBolt.java b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBolt.java index fd5f874..671e6b8 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBolt.java +++ b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBolt.java @@ -17,7 +17,7 @@ */ package org.apache.metron.enrichment.bolt; -import org.apache.metron.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetStrategy; import org.apache.storm.task.TopologyContext; import com.google.common.base.Joiner; import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig; diff --git a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBolt.java b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBolt.java index 55c7b66..f8a4223 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBolt.java +++ b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/GenericEnrichmentBolt.java @@ -28,18 +28,18 @@ import com.github.benmanes.caffeine.cache.CacheLoader; import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.LoadingCache; import org.apache.metron.common.Constants; -import org.apache.metron.common.bolt.ConfiguredEnrichmentBolt; +import org.apache.metron.storm.common.bolt.ConfiguredEnrichmentBolt; import org.apache.metron.common.configuration.ConfigurationType; import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig; import org.apache.metron.common.error.MetronError; import org.apache.metron.common.performance.PerformanceLogger; -import org.apache.metron.common.utils.ErrorUtils; import org.apache.metron.enrichment.cache.CacheKey; import org.apache.metron.enrichment.configuration.Enrichment; import org.apache.metron.enrichment.interfaces.EnrichmentAdapter; import org.apache.metron.enrichment.utils.EnrichmentUtils; import org.apache.metron.stellar.dsl.Context; import org.apache.metron.stellar.dsl.StellarFunctions; +import org.apache.metron.storm.common.utils.StormErrorUtils; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; @@ -214,7 +214,7 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt { .withErrorType(Constants.ErrorType.ENRICHMENT_ERROR) .withMessage("Unable to find SensorEnrichmentConfig for sourceType: " + sourceType) .addRawMessage(rawMessage); - ErrorUtils.handleError(collector, metronError); + StormErrorUtils.handleError(collector, metronError); continue; } config.getConfiguration().putIfAbsent(STELLAR_CONTEXT_CONF, stellarContext); @@ -239,7 +239,7 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt { .withThrowable(e) .withErrorFields(new HashSet() {{ add(field); }}) .addRawMessage(rawMessage); - ErrorUtils.handleError(collector, metronError); + StormErrorUtils.handleError(collector, metronError); continue; } } @@ -268,7 +268,7 @@ public class GenericEnrichmentBolt extends ConfiguredEnrichmentBolt { .withErrorType(Constants.ErrorType.ENRICHMENT_ERROR) .withThrowable(e) .addRawMessage(rawMessage); - ErrorUtils.handleError(collector, error); + StormErrorUtils.handleError(collector, error); } @Override diff --git a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/JoinBolt.java b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/JoinBolt.java index a9263fb..ac6a1cf 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/JoinBolt.java +++ b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/JoinBolt.java @@ -30,12 +30,12 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; import org.apache.metron.common.Constants; -import org.apache.metron.common.bolt.ConfiguredEnrichmentBolt; +import org.apache.metron.storm.common.bolt.ConfiguredEnrichmentBolt; 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.storm.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetters; import org.apache.metron.common.performance.PerformanceLogger; -import org.apache.metron.common.utils.ErrorUtils; +import org.apache.metron.storm.common.utils.StormErrorUtils; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; @@ -169,7 +169,7 @@ public abstract class JoinBolt<V> extends ConfiguredEnrichmentBolt { .withMessage("Joining problem: " + message) .withThrowable(e) .addRawMessage(message); - ErrorUtils.handleError(collector, error); + StormErrorUtils.handleError(collector, error); collector.ack(tuple); } perfLog.log("execute", "key={}, elapsed time to run execute", key); diff --git a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/SplitBolt.java b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/SplitBolt.java index de69ad4..48cec0b 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/SplitBolt.java +++ b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/SplitBolt.java @@ -20,7 +20,7 @@ package org.apache.metron.enrichment.bolt; import java.util.List; import java.util.Map; import java.util.Set; -import org.apache.metron.common.bolt.ConfiguredEnrichmentBolt; +import org.apache.metron.storm.common.bolt.ConfiguredEnrichmentBolt; import org.apache.metron.common.performance.PerformanceLogger; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; diff --git a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBolt.java b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBolt.java index 00c23ff..2b19375 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBolt.java +++ b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBolt.java @@ -23,7 +23,7 @@ import java.util.Map; import org.apache.metron.common.configuration.ConfigurationType; import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig; import org.apache.metron.common.configuration.enrichment.handler.ConfigHandler; -import org.apache.metron.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetStrategy; import org.apache.metron.common.utils.MessageUtils; import org.apache.metron.enrichment.adapters.maxmind.asn.GeoLiteAsnDatabase; import org.apache.metron.enrichment.adapters.maxmind.geo.GeoLiteCityDatabase; diff --git a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/UnifiedEnrichmentBolt.java b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/UnifiedEnrichmentBolt.java index d9f57eb..dd49de2 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/UnifiedEnrichmentBolt.java +++ b/metron-platform/metron-enrichment/metron-enrichment-storm/src/main/java/org/apache/metron/enrichment/bolt/UnifiedEnrichmentBolt.java @@ -20,14 +20,13 @@ package org.apache.metron.enrichment.bolt; import static org.apache.metron.common.Constants.STELLAR_CONTEXT_CONF; import org.apache.metron.common.Constants; -import org.apache.metron.common.bolt.ConfiguredEnrichmentBolt; +import org.apache.metron.storm.common.bolt.ConfiguredEnrichmentBolt; import org.apache.metron.common.configuration.ConfigurationType; import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig; 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.storm.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetters; import org.apache.metron.common.performance.PerformanceLogger; -import org.apache.metron.common.utils.ErrorUtils; import org.apache.metron.common.utils.MessageUtils; import org.apache.metron.enrichment.adapters.maxmind.asn.GeoLiteAsnDatabase; import org.apache.metron.enrichment.adapters.maxmind.geo.GeoLiteCityDatabase; @@ -41,6 +40,7 @@ import org.apache.metron.enrichment.parallel.ConcurrencyContext; import org.apache.metron.enrichment.parallel.WorkerPoolStrategies; import org.apache.metron.stellar.dsl.Context; import org.apache.metron.stellar.dsl.StellarFunctions; +import org.apache.metron.storm.common.utils.StormErrorUtils; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; @@ -285,7 +285,7 @@ public class UnifiedEnrichmentBolt extends ConfiguredEnrichmentBolt { .withMessage(t.getValue().getMessage()) .withThrowable(t.getValue()) .addRawMessage(t.getKey()); - ErrorUtils.handleError(collector, error); + StormErrorUtils.handleError(collector, error); } } catch (Exception e) { //If something terrible and unexpected happens then we want to send an error along, but this @@ -296,7 +296,7 @@ public class UnifiedEnrichmentBolt extends ConfiguredEnrichmentBolt { .withMessage(e.getMessage()) .withThrowable(e) .addRawMessage(message); - ErrorUtils.handleError(collector, error); + StormErrorUtils.handleError(collector, error); } finally { collector.ack(input); diff --git a/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBoltTest.java b/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBoltTest.java index 9fc8947..ed623f3 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBoltTest.java +++ b/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/EnrichmentJoinBoltTest.java @@ -26,7 +26,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; import org.adrianwalker.multilinestring.Multiline; -import org.apache.metron.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetStrategy; import org.apache.metron.test.bolt.BaseEnrichmentBoltTest; import org.apache.storm.tuple.Tuple; import org.json.simple.JSONObject; diff --git a/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java b/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java index bc7dace..8bdf409 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java +++ b/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/JoinBoltTest.java @@ -21,7 +21,7 @@ import com.github.benmanes.caffeine.cache.LoadingCache; import org.adrianwalker.multilinestring.Multiline; import org.apache.metron.common.Constants; import org.apache.metron.common.error.MetronError; -import org.apache.metron.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetStrategy; import org.apache.metron.test.bolt.BaseEnrichmentBoltTest; import org.apache.metron.test.error.MetronErrorJSONMatcher; import org.apache.storm.task.TopologyContext; diff --git a/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBoltTest.java b/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBoltTest.java index a05c843..aeafede 100644 --- a/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBoltTest.java +++ b/metron-platform/metron-enrichment/metron-enrichment-storm/src/test/java/org/apache/metron/enrichment/bolt/ThreatIntelJoinBoltTest.java @@ -31,7 +31,7 @@ import java.util.Map; import org.adrianwalker.multilinestring.Multiline; import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig; import org.apache.metron.common.configuration.enrichment.threatintel.ThreatTriageConfig; -import org.apache.metron.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetStrategy; import org.apache.metron.common.utils.JSONUtils; import org.apache.metron.enrichment.adapters.maxmind.asn.GeoLiteAsnDatabase; import org.apache.metron.enrichment.adapters.maxmind.geo.GeoLiteCityDatabase; diff --git a/metron-platform/metron-parsing/metron-parsing-storm/pom.xml b/metron-platform/metron-parsing/metron-parsing-storm/pom.xml index 72770c8..c936ef9 100644 --- a/metron-platform/metron-parsing/metron-parsing-storm/pom.xml +++ b/metron-platform/metron-parsing/metron-parsing-storm/pom.xml @@ -33,7 +33,7 @@ <!-- Metron dependencies --> <dependency> <groupId>org.apache.metron</groupId> - <artifactId>metron-common</artifactId> + <artifactId>metron-common-storm</artifactId> <version>${project.parent.version}</version> </dependency> <dependency> diff --git a/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java b/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java index 21aa087..2837fcb 100644 --- a/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java +++ b/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/ParserBolt.java @@ -30,18 +30,18 @@ import java.util.function.Function; import java.util.stream.Collectors; import org.apache.metron.common.Constants; -import org.apache.metron.common.bolt.ConfiguredParserBolt; +import org.apache.metron.storm.common.bolt.ConfiguredParserBolt; import org.apache.metron.common.configuration.ParserConfigurations; import org.apache.metron.common.configuration.SensorParserConfig; 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.message.MessageGetters; +import org.apache.metron.storm.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetters; import org.apache.metron.common.message.metadata.RawMessage; -import org.apache.metron.common.message.metadata.RawMessageUtil; -import org.apache.metron.common.utils.ErrorUtils; +import org.apache.metron.storm.common.message.metadata.RawMessageUtil; import org.apache.metron.common.utils.MessageUtils; import org.apache.metron.common.writer.BulkMessage; +import org.apache.metron.storm.common.utils.StormErrorUtils; import org.apache.metron.writer.AckTuplesPolicy; import org.apache.metron.parsers.ParserRunner; import org.apache.metron.parsers.ParserRunnerResults; @@ -255,7 +255,7 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable { , sensorParserConfig.getRawMessageStrategyConfig() ); ParserRunnerResults<JSONObject> parserRunnerResults = parserRunner.execute(sensorType, rawMessage, parserConfigurations); - parserRunnerResults.getErrors().forEach(error -> ErrorUtils.handleError(collector, error)); + parserRunnerResults.getErrors().forEach(error -> StormErrorUtils.handleError(collector, error)); WriterHandler writer = sensorToWriterMap.get(sensorType); int numWritten = 0; @@ -326,7 +326,7 @@ public class ParserBolt extends ConfiguredParserBolt implements Serializable { .withThrowable(ex) .withSensorType(Collections.singleton(sensorType)) .addRawMessage(originalMessage); - ErrorUtils.handleError(collector, error); + StormErrorUtils.handleError(collector, error); } @Override diff --git a/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java b/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java index 6a174f3..1ee6464 100644 --- a/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java +++ b/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/WriterBolt.java @@ -21,11 +21,11 @@ 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.metron.storm.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetters; import org.apache.metron.common.utils.MessageUtils; import org.apache.metron.common.writer.BulkMessage; +import org.apache.metron.storm.common.utils.StormErrorUtils; import org.apache.metron.writer.AckTuplesPolicy; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; @@ -94,7 +94,7 @@ public class WriterBolt extends BaseRichBolt { .withThrowable(e) .withSensorType(Collections.singleton(sensorType)) .addRawMessage(message); - ErrorUtils.handleError(collector, error); + StormErrorUtils.handleError(collector, error); collector.ack(tuple); } } diff --git a/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/WriterHandler.java b/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/WriterHandler.java index 434db45..4f5ebe9 100644 --- a/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/WriterHandler.java +++ b/metron-platform/metron-parsing/metron-parsing-storm/src/main/java/org/apache/metron/parsers/bolt/WriterHandler.java @@ -27,7 +27,7 @@ import org.apache.metron.common.configuration.writer.ConfigurationStrategy; import org.apache.metron.common.configuration.writer.ConfigurationsStrategies; import org.apache.metron.common.configuration.writer.SingleBatchConfigurationFacade; import org.apache.metron.common.configuration.writer.WriterConfiguration; -import org.apache.metron.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetStrategy; import org.apache.metron.common.writer.BulkMessageWriter; import org.apache.metron.common.writer.BulkMessage; import org.apache.metron.common.writer.MessageWriter; @@ -75,7 +75,7 @@ public class WriterHandler implements Serializable { writerTransformer = config -> new SingleBatchConfigurationFacade(configStrategy.createWriterConfig(messageWriter, config)); } try { - messageWriter.init(stormConf, topologyContext, writerTransformer.apply(configurations)); + messageWriter.init(stormConf, writerTransformer.apply(configurations)); } catch (Exception e) { throw new IllegalStateException("Unable to initialize message writer", e); } diff --git a/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java b/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java index 90c882f..4240f7a 100644 --- a/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java +++ b/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/bolt/ParserBoltTest.java @@ -22,7 +22,7 @@ import org.apache.metron.common.configuration.IndexingConfigurations; import org.apache.metron.common.configuration.ParserConfigurations; import org.apache.metron.common.configuration.SensorParserConfig; import org.apache.metron.common.error.MetronError; -import org.apache.metron.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetStrategy; import org.apache.metron.common.message.metadata.RawMessage; import org.apache.metron.common.writer.BulkMessage; import org.apache.metron.writer.AckTuplesPolicy; diff --git a/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java b/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java index 68fc15f..e16c5e9 100644 --- a/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java +++ b/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/bolt/WriterBoltTest.java @@ -106,7 +106,7 @@ public class WriterBoltTest extends BaseBoltTest{ } bolt.prepare(new HashMap(), topologyContext, outputCollector); - verify(batchWriter, times(1)).init(any(), any(), any()); + verify(batchWriter, times(1)).init(any(), any()); for(int i = 0;i < 4;++i) { Tuple t = tuples.get(i); bolt.execute(t); @@ -206,7 +206,7 @@ public class WriterBoltTest extends BaseBoltTest{ bolt.prepare(new HashMap(), topologyContext, outputCollector); - verify(batchWriter, times(1)).init(any(), any(), any()); + verify(batchWriter, times(1)).init(any(), any()); for(int i = 0;i < 4;++i) { Tuple t = tuples.get(i); @@ -264,7 +264,7 @@ public class WriterBoltTest extends BaseBoltTest{ when(errorTuple.getValueByField(eq("message"))).thenReturn(errorMessage); bolt.prepare(new HashMap(), topologyContext, outputCollector); - verify(batchWriter, times(1)).init(any(), any(), any()); + verify(batchWriter, times(1)).init(any(), any()); for(int i = 0;i < 4;++i) { Tuple t = tuples.get(i); @@ -316,7 +316,7 @@ public class WriterBoltTest extends BaseBoltTest{ bolt.prepare(new HashMap(), topologyContext, outputCollector); doThrow(new Exception()).when(batchWriter).write(any(), any(), any()); - verify(batchWriter, times(1)).init(any(), any(), any()); + verify(batchWriter, times(1)).init(any(), any()); for(int i = 0;i < 4;++i) { Tuple t = tuples.get(i); bolt.execute(t); diff --git a/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/integration/validation/StormParserDriver.java b/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/integration/validation/StormParserDriver.java index 66b6c2b..0d46d73 100644 --- a/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/integration/validation/StormParserDriver.java +++ b/metron-platform/metron-parsing/metron-parsing-storm/src/test/java/org/apache/metron/parsers/integration/validation/StormParserDriver.java @@ -60,7 +60,7 @@ public class StormParserDriver extends ParserDriver { } @Override - public void init(Map stormConf, TopologyContext topologyContext, WriterConfiguration config) throws Exception { + public void init(Map stormConf, WriterConfiguration config) throws Exception { } diff --git a/metron-platform/metron-solr/src/main/java/org/apache/metron/solr/writer/SolrWriter.java b/metron-platform/metron-solr/src/main/java/org/apache/metron/solr/writer/SolrWriter.java index 6fb2b35..b23a517 100644 --- a/metron-platform/metron-solr/src/main/java/org/apache/metron/solr/writer/SolrWriter.java +++ b/metron-platform/metron-solr/src/main/java/org/apache/metron/solr/writer/SolrWriter.java @@ -50,7 +50,6 @@ import org.apache.solr.client.solrj.impl.Krb5HttpClientConfigurer; import org.apache.solr.client.solrj.response.UpdateResponse; import org.apache.solr.common.SolrException; import org.apache.solr.common.SolrInputDocument; -import org.apache.storm.task.TopologyContext; import org.json.simple.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -153,7 +152,7 @@ public class SolrWriter implements BulkMessageWriter<JSONObject>, Serializable { } @Override - public void init(Map stormConf, TopologyContext topologyContext, WriterConfiguration configurations) throws IOException, SolrServerException { + public void init(Map stormConf, WriterConfiguration configurations) throws IOException, SolrServerException { Map<String, Object> globalConfiguration = configurations.getGlobalConfig(); initializeFromGlobalConfig(globalConfiguration); LOG.info("Initializing SOLR writer: {}", zookeeperUrl); diff --git a/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/integration/schema/SchemaValidationIntegrationTest.java b/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/integration/schema/SchemaValidationIntegrationTest.java index a4908ad..ddc7801 100644 --- a/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/integration/schema/SchemaValidationIntegrationTest.java +++ b/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/integration/schema/SchemaValidationIntegrationTest.java @@ -157,7 +157,7 @@ public class SchemaValidationIntegrationTest { } }; - solrWriter.init(null, null, writerConfig); + solrWriter.init(null, writerConfig); BulkWriterResponse response = solrWriter.write(sensorType, writerConfig, messages); Assert.assertTrue(response.getErrors().isEmpty()); diff --git a/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/writer/SolrWriterTest.java b/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/writer/SolrWriterTest.java index 0f8dab1..515f2f8 100644 --- a/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/writer/SolrWriterTest.java +++ b/metron-platform/metron-solr/src/test/java/org/apache/metron/solr/writer/SolrWriterTest.java @@ -121,7 +121,7 @@ public class SolrWriterTest { String collection = "metron"; MetronSolrClient solr = Mockito.mock(MetronSolrClient.class); SolrWriter writer = new SolrWriter().withMetronSolrClient(solr); - writer.init(null, null,new IndexingWriterConfiguration("solr", configurations)); + writer.init(null,new IndexingWriterConfiguration("solr", configurations)); verify(solr, times(1)).setDefaultCollection(collection); collection = "metron2"; @@ -129,7 +129,7 @@ public class SolrWriterTest { globalConfig.put("solr.collection", collection); configurations.updateGlobalConfig(globalConfig); writer = new SolrWriter().withMetronSolrClient(solr); - writer.init(null, null, new IndexingWriterConfiguration("solr", configurations)); + writer.init(null, new IndexingWriterConfiguration("solr", configurations)); verify(solr, times(1)).setDefaultCollection(collection); writer.write("test", new IndexingWriterConfiguration("solr", configurations), messages); diff --git a/metron-platform/metron-writer/pom.xml b/metron-platform/metron-writer/pom.xml index 2376bff..2205cec 100644 --- a/metron-platform/metron-writer/pom.xml +++ b/metron-platform/metron-writer/pom.xml @@ -219,7 +219,7 @@ </dependency> <dependency> <groupId>org.apache.metron</groupId> - <artifactId>metron-common</artifactId> + <artifactId>metron-common-storm</artifactId> <version>${project.parent.version}</version> </dependency> <dependency> diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/AckTuplesPolicy.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/AckTuplesPolicy.java index 64685bf..2543309 100644 --- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/AckTuplesPolicy.java +++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/AckTuplesPolicy.java @@ -20,7 +20,7 @@ package org.apache.metron.writer; 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.storm.common.message.MessageGetStrategy; import org.apache.metron.common.writer.BulkMessage; import org.apache.metron.common.writer.BulkWriterResponse; import org.apache.metron.common.writer.MessageId; diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/NoopWriter.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/NoopWriter.java index 01def5d..c81076a 100644 --- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/NoopWriter.java +++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/NoopWriter.java @@ -19,7 +19,6 @@ package org.apache.metron.writer; import org.apache.metron.common.writer.BulkMessage; import org.apache.metron.common.writer.MessageId; -import org.apache.storm.task.TopologyContext; import com.google.common.base.Splitter; import com.google.common.collect.Iterables; import org.apache.metron.common.configuration.writer.WriterConfiguration; @@ -130,7 +129,7 @@ public class NoopWriter extends AbstractWriter implements BulkMessageWriter<JSON } @Override - public void init(Map stormConf, TopologyContext topologyContext, WriterConfiguration config) throws Exception { + public void init(Map stormConf, WriterConfiguration config) throws Exception { } @Override diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/WriterToBulkWriter.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/WriterToBulkWriter.java index 709875a..6b8bb2f 100644 --- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/WriterToBulkWriter.java +++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/WriterToBulkWriter.java @@ -20,7 +20,6 @@ package org.apache.metron.writer; import org.apache.metron.common.writer.BulkMessage; import org.apache.metron.common.writer.MessageId; -import org.apache.storm.task.TopologyContext; import com.google.common.collect.Iterables; import org.apache.metron.common.configuration.writer.SingleBatchConfigurationFacade; import org.apache.metron.common.configuration.writer.WriterConfiguration; @@ -44,7 +43,7 @@ public class WriterToBulkWriter<MESSAGE_T> implements BulkMessageWriter<MESSAGE_ this.messageWriter = messageWriter; } @Override - public void init(Map stormConf, TopologyContext topologyContext, WriterConfiguration config) throws Exception { + public void init(Map stormConf, WriterConfiguration config) throws Exception { messageWriter.init(); } 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 c9215e3..8d38c60 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 @@ -29,21 +29,22 @@ import java.util.function.Function; import com.google.common.collect.Iterables; import org.apache.metron.common.Constants; -import org.apache.metron.common.bolt.ConfiguredBolt; +import org.apache.metron.storm.common.bolt.ConfiguredBolt; import org.apache.metron.common.configuration.Configurations; 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.message.MessageGetters; +import org.apache.metron.storm.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetters; import org.apache.metron.common.system.Clock; -import org.apache.metron.common.utils.ErrorUtils; import org.apache.metron.common.utils.MessageUtils; import org.apache.metron.common.writer.BulkMessageWriter; import org.apache.metron.common.writer.BulkMessage; import org.apache.metron.common.writer.MessageWriter; +import org.apache.metron.storm.common.utils.StormErrorUtils; import org.apache.metron.writer.AckTuplesPolicy; import org.apache.metron.writer.BulkWriterComponent; import org.apache.metron.writer.WriterToBulkWriter; +import org.apache.metron.writer.hdfs.HdfsWriter; import org.apache.storm.Config; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; @@ -235,7 +236,10 @@ public class BulkMessageWriterBolt<CONFIG_T extends Configurations> extends Conf BulkWriterComponent<JSONObject> bulkWriterComponent = new BulkWriterComponent<>(maxBatchTimeout); bulkWriterComponent.addFlushPolicy(ackTuplesPolicy); setWriterComponent(bulkWriterComponent); - bulkMessageWriter.init(stormConf, context, writerconf); + bulkMessageWriter.init(stormConf, writerconf); + if (bulkMessageWriter instanceof HdfsWriter) { + ((HdfsWriter) bulkMessageWriter).initFileNameFormat(context); + } } catch (Exception e) { throw new RuntimeException(e); } @@ -362,7 +366,7 @@ public class BulkMessageWriterBolt<CONFIG_T extends Configurations> extends Conf .withErrorType(Constants.ErrorType.INDEXING_ERROR) .withThrowable(e); collector.ack(tuple); - ErrorUtils.handleError(collector, error); + StormErrorUtils.handleError(collector, error); } @Override diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/hbase/SimpleHbaseEnrichmentWriter.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/hbase/SimpleHbaseEnrichmentWriter.java index 50b11e1..d7b2cb9 100644 --- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/hbase/SimpleHbaseEnrichmentWriter.java +++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/hbase/SimpleHbaseEnrichmentWriter.java @@ -20,7 +20,6 @@ package org.apache.metron.writer.hbase; import org.apache.metron.common.writer.BulkMessage; import org.apache.metron.common.writer.MessageId; -import org.apache.storm.task.TopologyContext; import com.google.common.collect.ImmutableList; import com.google.common.collect.Iterables; import org.apache.hadoop.conf.Configuration; @@ -178,7 +177,7 @@ public class SimpleHbaseEnrichmentWriter extends AbstractWriter implements BulkM } @Override - public void init(Map stormConf, TopologyContext topologyContext, WriterConfiguration configuration) throws Exception { + public void init(Map stormConf, WriterConfiguration configuration) throws Exception { if(converter == null) { converter = new EnrichmentConverter(); } diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/hdfs/HdfsWriter.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/hdfs/HdfsWriter.java index 9e6827b..d251602 100644 --- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/hdfs/HdfsWriter.java +++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/hdfs/HdfsWriter.java @@ -88,10 +88,9 @@ public class HdfsWriter implements BulkMessageWriter<JSONObject>, Serializable { } @Override - public void init(Map stormConfig, TopologyContext topologyContext, WriterConfiguration configurations) { + public void init(Map stormConfig, WriterConfiguration configurations) { this.stormConfig = stormConfig; this.stellarProcessor = new StellarProcessor(); - this.fileNameFormat.prepare(stormConfig,topologyContext); if(syncPolicy != null) { //if the user has specified the sync policy, we don't want to override their wishes. LOG.debug("Using user specified sync policy {}", syncPolicy.getClass().getSimpleName()); @@ -104,6 +103,10 @@ public class HdfsWriter implements BulkMessageWriter<JSONObject>, Serializable { } } + public void initFileNameFormat(TopologyContext topologyContext) { + this.fileNameFormat.prepare(stormConfig,topologyContext); + } + @Override public BulkWriterResponse write(String sensorType, WriterConfiguration configurations, List<BulkMessage<JSONObject>> messages) throws Exception { BulkWriterResponse response = new BulkWriterResponse(); diff --git a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/kafka/KafkaWriter.java b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/kafka/KafkaWriter.java index 78a27fd..a313057 100644 --- a/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/kafka/KafkaWriter.java +++ b/metron-platform/metron-writer/src/main/java/org/apache/metron/writer/kafka/KafkaWriter.java @@ -44,7 +44,6 @@ import org.apache.metron.common.writer.BulkWriterResponse; import org.apache.metron.common.writer.MessageId; import org.apache.metron.stellar.common.utils.ConversionUtils; import org.apache.metron.writer.AbstractWriter; -import org.apache.storm.task.TopologyContext; import org.json.simple.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -200,7 +199,7 @@ public class KafkaWriter extends AbstractWriter implements BulkMessageWriter<JSO } @Override - public void init(Map stormConf, TopologyContext topologyContext, WriterConfiguration config) + public void init(Map stormConf, WriterConfiguration config) throws Exception { if(this.zkQuorum != null && this.brokerUrl == null) { try { diff --git a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/AckTuplesPolicyTest.java b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/AckTuplesPolicyTest.java index 1d970cb..04efb3d 100644 --- a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/AckTuplesPolicyTest.java +++ b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/AckTuplesPolicyTest.java @@ -19,7 +19,7 @@ package org.apache.metron.writer; import org.apache.metron.common.Constants; import org.apache.metron.common.error.MetronError; -import org.apache.metron.common.message.MessageGetStrategy; +import org.apache.metron.storm.common.message.MessageGetStrategy; import org.apache.metron.common.writer.BulkWriterResponse; import org.apache.metron.common.writer.MessageId; import org.apache.metron.test.error.MetronErrorJSONMatcher; diff --git a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/BulkWriterComponentTest.java b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/BulkWriterComponentTest.java index 16f3b4f..e94608f 100644 --- a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/BulkWriterComponentTest.java +++ b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/BulkWriterComponentTest.java @@ -18,7 +18,6 @@ package org.apache.metron.writer; 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.reset; diff --git a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/bolt/BulkMessageWriterBoltTest.java b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/bolt/BulkMessageWriterBoltTest.java index df80296..0fd784f 100644 --- a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/bolt/BulkMessageWriterBoltTest.java +++ b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/bolt/BulkMessageWriterBoltTest.java @@ -42,7 +42,7 @@ import org.apache.log4j.Level; import org.apache.metron.common.Constants; import org.apache.metron.common.configuration.IndexingConfigurations; import org.apache.metron.common.configuration.writer.WriterConfiguration; -import org.apache.metron.common.message.MessageGetters; +import org.apache.metron.storm.common.message.MessageGetters; import org.apache.metron.common.system.FakeClock; import org.apache.metron.common.writer.BulkMessageWriter; import org.apache.metron.common.writer.BulkMessage; @@ -51,7 +51,6 @@ import org.apache.metron.common.writer.MessageId; import org.apache.metron.test.bolt.BaseEnrichmentBoltTest; import org.apache.metron.test.utils.UnitTestHelper; import org.apache.metron.writer.BulkWriterComponent; -import org.apache.storm.task.TopologyContext; import org.apache.storm.tuple.Tuple; import org.apache.storm.tuple.Values; import org.json.simple.JSONObject; @@ -163,7 +162,7 @@ public class BulkMessageWriterBoltTest extends BaseEnrichmentBoltTest { bulkMessageWriterBolt.getConfigurations().updateSensorIndexingConfig(sensorType, new FileInputStream(sampleSensorIndexingConfigPath)); { - doThrow(new Exception()).when(bulkMessageWriter).init(eq(stormConf),any(TopologyContext.class), any(WriterConfiguration.class)); + doThrow(new Exception()).when(bulkMessageWriter).init(eq(stormConf), any(WriterConfiguration.class)); try { bulkMessageWriterBolt.prepare(stormConf, topologyContext, outputCollector); fail("A runtime exception should be thrown when bulkMessageWriter.init throws an exception"); @@ -173,7 +172,7 @@ public class BulkMessageWriterBoltTest extends BaseEnrichmentBoltTest { { when(bulkMessageWriter.getName()).thenReturn("hdfs"); bulkMessageWriterBolt.prepare(stormConf, topologyContext, outputCollector); - verify(bulkMessageWriter, times(1)).init(eq(stormConf),any(TopologyContext.class), any(WriterConfiguration.class)); + verify(bulkMessageWriter, times(1)).init(eq(stormConf), any(WriterConfiguration.class)); } { for(int i = 0; i < 4; i++) { @@ -222,7 +221,7 @@ public class BulkMessageWriterBoltTest extends BaseEnrichmentBoltTest { Map stormConf = new HashMap(); when(bulkMessageWriter.getName()).thenReturn("elasticsearch"); bulkMessageWriterBolt.prepare(stormConf, topologyContext, outputCollector, clock); - verify(bulkMessageWriter, times(1)).init(eq(stormConf),any(TopologyContext.class), any(WriterConfiguration.class)); + verify(bulkMessageWriter, times(1)).init(eq(stormConf), any(WriterConfiguration.class)); } { int batchTimeout = bulkMessageWriterBolt.getMaxBatchTimeout(); @@ -260,8 +259,7 @@ public class BulkMessageWriterBoltTest extends BaseEnrichmentBoltTest { Map stormConf = new HashMap(); when(bulkMessageWriter.getName()).thenReturn("elasticsearch"); bulkMessageWriterBolt.prepare(stormConf, topologyContext, outputCollector, clock); - verify(bulkMessageWriter, times(1)).init(eq(stormConf),any(TopologyContext.class) - , any(WriterConfiguration.class)); + verify(bulkMessageWriter, times(1)).init(eq(stormConf), any(WriterConfiguration.class)); } { int batchTimeout = bulkMessageWriterBolt.getMaxBatchTimeout(); diff --git a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/hdfs/HdfsWriterTest.java b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/hdfs/HdfsWriterTest.java index 88ba4fd..e48c8e2 100644 --- a/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/hdfs/HdfsWriterTest.java +++ b/metron-platform/metron-writer/src/test/java/org/apache/metron/writer/hdfs/HdfsWriterTest.java @@ -76,7 +76,8 @@ public class HdfsWriterTest { public void testGetHdfsPathNull() { WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, new IndexingConfigurations()); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(),createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); Object result = writer.getHdfsPathExtension(SENSOR_NAME,null, message); @@ -89,7 +90,8 @@ public class HdfsWriterTest { public void testGetHdfsPathEmptyString() { WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, new IndexingConfigurations()); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); Object result = writer.getHdfsPathExtension(SENSOR_NAME, "", message); @@ -102,7 +104,8 @@ public class HdfsWriterTest { public void testGetHdfsPathConstant() { WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, new IndexingConfigurations()); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); Object result = writer.getHdfsPathExtension(SENSOR_NAME, "'new'", message); @@ -115,7 +118,8 @@ public class HdfsWriterTest { public void testGetHdfsPathDirectVariable() { WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, new IndexingConfigurations()); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); message.put("test.key", "test.value"); @@ -129,7 +133,8 @@ public class HdfsWriterTest { public void testGetHdfsPathFormatConstant() { WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, new IndexingConfigurations()); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); Object result = writer.getHdfsPathExtension(SENSOR_NAME, "FORMAT('/test/folder/')", message); @@ -143,7 +148,8 @@ public class HdfsWriterTest { IndexingConfigurations indexingConfig = new IndexingConfigurations(); WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, indexingConfig); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); message.put("test.key", "test.value"); @@ -159,7 +165,8 @@ public class HdfsWriterTest { IndexingConfigurations indexingConfig = new IndexingConfigurations(); WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, indexingConfig); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); String filename = writer.fileNameFormat.getName(1,1); Assert.assertEquals("prefix-Xcom-7-1-1.json", filename); writer.close(); @@ -171,7 +178,8 @@ public class HdfsWriterTest { IndexingConfigurations indexingConfig = new IndexingConfigurations(); WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, indexingConfig); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); message.put("test.key", "test.value"); @@ -193,7 +201,8 @@ public class HdfsWriterTest { IndexingConfigurations indexingConfig = new IndexingConfigurations(); WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, indexingConfig); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); message.put("test.key", "test.value"); @@ -206,7 +215,8 @@ public class HdfsWriterTest { public void testGetHdfsPathNonString() { WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, new IndexingConfigurations()); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); writer.getHdfsPathExtension(SENSOR_NAME, "{'key':'value'}", message); @@ -220,7 +230,8 @@ public class HdfsWriterTest { WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, indexingConfig); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat) .withMaxOpenFiles(maxFiles); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); for(int i = 0; i < maxFiles; i++) { writer.getSourceHandler(SENSOR_NAME, Integer.toString(i), null); @@ -235,7 +246,8 @@ public class HdfsWriterTest { WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, indexingConfig); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat) .withMaxOpenFiles(maxFiles); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); for(int i = 0; i < maxFiles+1; i++) { writer.getSourceHandler(SENSOR_NAME, Integer.toString(i), null); @@ -253,7 +265,8 @@ public class HdfsWriterTest { HdfsWriter writer = new HdfsWriter().withFileNameFormat(format); IndexingConfigurations indexingConfig = new IndexingConfigurations(); WriterConfiguration config = new IndexingWriterConfiguration(WRITER_NAME, indexingConfig); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); message.put("test.key", "test.value"); @@ -296,7 +309,8 @@ public class HdfsWriterTest { .withExtension(".json") .withPrefix("prefix-"); HdfsWriter writer = new HdfsWriter().withFileNameFormat(format); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); // These two messages will be routed to the same folder, because test.key is the same JSONObject message = new JSONObject(); @@ -339,7 +353,8 @@ public class HdfsWriterTest { .withExtension(".json") .withPrefix("prefix-"); HdfsWriter writer = new HdfsWriter().withFileNameFormat(format); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); // These two messages will be routed to the same folder, because test.key is the same JSONObject message = new JSONObject(); @@ -395,7 +410,8 @@ public class HdfsWriterTest { .withExtension(".json") .withPrefix("prefix-"); HdfsWriter writer = new HdfsWriter().withFileNameFormat(format); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); // These two messages will be routed to the same folder, because test.key is the same JSONObject message = new JSONObject(); @@ -428,7 +444,8 @@ public class HdfsWriterTest { String function = "FORMAT('test-%s/%s', test.key, test.key)"; WriterConfiguration config = buildWriterConfiguration(function); HdfsWriter writer = new HdfsWriter().withFileNameFormat(testFormat); - writer.init(new HashMap<String, String>(), createTopologyContext(), config); + writer.init(new HashMap<String, String>(), config); + writer.initFileNameFormat(createTopologyContext()); JSONObject message = new JSONObject(); message.put("test.key", "test.value"); diff --git a/metron-platform/pom.xml b/metron-platform/pom.xml index af563cf..5fd4f29 100644 --- a/metron-platform/pom.xml +++ b/metron-platform/pom.xml @@ -62,6 +62,7 @@ <module>metron-zookeeper</module> <module>metron-parsing</module> <module>metron-hbase-server</module> + <module>metron-common-streaming</module> </modules> <dependencies> <dependency>