diff --git a/DEPENDENCY-LICENSES b/DEPENDENCY-LICENSES index b38f864ef30..2256de21e3d 100644 --- a/DEPENDENCY-LICENSES +++ b/DEPENDENCY-LICENSES @@ -85,13 +85,11 @@ List of third-party dependencies grouped by their license type. * Apache HBase Relocated (Shaded) Third-party Miscellaneous Libs (org.apache.hbase.thirdparty:hbase-shaded-miscellaneous:4.1.9 - https://hbase.apache.org/hbase-shaded-miscellaneous) * Apache HBase - Shaded Protocol (org.apache.hbase:hbase-protocol-shaded:2.6.1-hadoop3 - https://hbase.apache.org/hbase-build-configuration/hbase-protocol-shaded) * Apache HBase Unsafe Wrapper (org.apache.hbase.thirdparty:hbase-unsafe:4.1.9 - https://hbase.apache.org/hbase-unsafe) - * Apache HttpAsyncClient (org.apache.httpcomponents:httpasyncclient:4.1.5 - http://hc.apache.org/httpcomponents-asyncclient) * Apache HttpClient (org.apache.httpcomponents:httpclient:4.5.14 - http://hc.apache.org/httpcomponents-client-ga) * Apache HttpClient (org.apache.httpcomponents.client5:httpclient5:5.2.1 - https://hc.apache.org/httpcomponents-client-5.0.x/5.2.1/httpclient5/) * Apache HttpComponents Core HTTP/1.1 (org.apache.httpcomponents.core5:httpcore5:5.2 - https://hc.apache.org/httpcomponents-core-5.2.x/5.2/httpcore5/) * Apache HttpComponents Core HTTP/2 (org.apache.httpcomponents.core5:httpcore5-h2:5.2 - https://hc.apache.org/httpcomponents-core-5.2.x/5.2/httpcore5-h2/) * Apache HttpCore (org.apache.httpcomponents:httpcore:4.4.16 - http://hc.apache.org/httpcomponents-core-ga) - * Apache HttpCore NIO (org.apache.httpcomponents:httpcore-nio:4.4.15 - http://hc.apache.org/httpcomponents-core-ga) * Apache Ivy (org.apache.ivy:ivy:2.4.0 - http://ant.apache.org/ivy/) * Apache Kafka (org.apache.kafka:kafka-clients:3.9.0 - https://kafka.apache.org) * Apache Log4j 1.x Compatibility API (org.apache.logging.log4j:log4j-1.2-api:2.24.1 - https://logging.apache.org/log4j/2.x/log4j/log4j-1.2-api/) @@ -136,7 +134,6 @@ List of third-party dependencies grouped by their license type. * Commons Lang (commons-lang:commons-lang:2.6 - http://commons.apache.org/lang/) * Commons Logging (commons-logging:commons-logging:1.1.3 - http://commons.apache.org/proper/commons-logging/) * Commons Pool (commons-pool:commons-pool:1.5.4 - http://commons.apache.org/pool/) - * compiler (com.github.spullara.mustache.java:compiler:0.9.6 - http://github.com/spullara/mustache.java) * Curator Client (org.apache.curator:curator-client:5.7.1 - https://curator.apache.org/curator-client) * Curator Framework (org.apache.curator:curator-framework:5.7.1 - https://curator.apache.org/curator-framework) * Curator Recipes (org.apache.curator:curator-recipes:5.7.1 - https://curator.apache.org/curator-recipes) @@ -202,13 +199,11 @@ List of third-party dependencies grouped by their license type. * Hive Upgrade Acid (org.apache.hive:hive-upgrade-acid:3.1.3 - https://www.apache.org/hive-upgrade-acid/) * Hive Vector-Code-Gen Utilities (org.apache.hive:hive-vector-code-gen:3.1.3 - https://hive.apache.org/hive-vector-code-gen) * HPPC Collections (com.carrotsearch:hppc:0.7.2 - http://labs.carrotsearch.com/hppc.html/hppc) - * HPPC Collections (com.carrotsearch:hppc:0.8.1 - http://labs.carrotsearch.com/hppc.html/hppc) * j2html (com.j2html:j2html:1.6.0 - http://j2html.com) * J2ObjC Annotations (com.google.j2objc:j2objc-annotations:3.0.0 - https://github.com/google/j2objc/) * Jackson-annotations (com.fasterxml.jackson.core:jackson-annotations:2.18.1 - https://github.com/FasterXML/jackson) * Jackson-core (com.fasterxml.jackson.core:jackson-core:2.18.1 - https://github.com/FasterXML/jackson-core) * jackson-databind (com.fasterxml.jackson.core:jackson-databind:2.18.1 - https://github.com/FasterXML/jackson) - * Jackson dataformat: CBOR (com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.18.1 - https://github.com/FasterXML/jackson-dataformats-binary) * Jackson dataformat: Smile (com.fasterxml.jackson.dataformat:jackson-dataformat-smile:2.18.1 - https://github.com/FasterXML/jackson-dataformats-binary) * Jackson-dataformat-YAML (com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.18.1 - https://github.com/FasterXML/jackson-dataformats-text) * Jackson datatype: Guava (com.fasterxml.jackson.datatype:jackson-datatype-guava:2.18.1 - https://github.com/FasterXML/jackson-datatypes-collections) @@ -248,19 +243,6 @@ List of third-party dependencies grouped by their license type. * Kerby PKIX Project (org.apache.kerby:kerby-pkix:2.0.3 - https://directory.apache.org/kerby/kerby-pkix) * Kerby Util (org.apache.kerby:kerby-util:2.0.3 - https://directory.apache.org/kerby/kerby-common/kerby-util) * Logging (commons-logging:commons-logging:1.0.3 - http://jakarta.apache.org/commons/logging/) - * Lucene Common Analyzers (org.apache.lucene:lucene-analyzers-common:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-analyzers-common) - * Lucene Core (org.apache.lucene:lucene-core:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-core) - * Lucene Grouping (org.apache.lucene:lucene-grouping:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-grouping) - * Lucene Highlighter (org.apache.lucene:lucene-highlighter:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-highlighter) - * Lucene Join (org.apache.lucene:lucene-join:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-join) - * Lucene Memory (org.apache.lucene:lucene-backward-codecs:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-backward-codecs) - * Lucene Memory (org.apache.lucene:lucene-memory:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-memory) - * Lucene Miscellaneous (org.apache.lucene:lucene-misc:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-misc) - * Lucene Queries (org.apache.lucene:lucene-queries:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-queries) - * Lucene QueryParsers (org.apache.lucene:lucene-queryparser:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-queryparser) - * Lucene Sandbox (org.apache.lucene:lucene-sandbox:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-sandbox) - * Lucene Spatial 3D (org.apache.lucene:lucene-spatial3d:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-spatial3d) - * Lucene Suggest (org.apache.lucene:lucene-suggest:8.11.1 - https://lucene.apache.org/lucene-parent/lucene-suggest) * LZ4 and xxHash (org.lz4:lz4-java:1.8.0 - https://github.com/lz4/lz4-java) * Maven Artifact (org.apache.maven:maven-artifact:3.6.0 - https://maven.apache.org/ref/3.6.0/maven-artifact/) * Maven Artifact (org.apache.maven:maven-artifact:3.8.1 - https://maven.apache.org/ref/3.8.1/maven-artifact/) @@ -359,7 +341,6 @@ List of third-party dependencies grouped by their license type. * Prometheus Metrics Tracer Initializer (io.prometheus:prometheus-metrics-tracer-initializer:1.3.3 - http://github.com/prometheus/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-initializer) * Prometheus Metrics Tracer OpenTelemetry (io.prometheus:prometheus-metrics-tracer-otel:1.3.3 - http://github.com/prometheus/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-otel) * Prometheus Metrics Tracer OpenTelemetry Agent (io.prometheus:prometheus-metrics-tracer-otel-agent:1.3.3 - http://github.com/prometheus/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-otel-agent) - * rest (org.elasticsearch.client:elasticsearch-rest-client:7.17.13 - https://github.com/elastic/elasticsearch.git) * sigar (org.fusesource:sigar:1.6.4 - http://fusesource.com/sigar/) * SLF4J 1 Binding for Log4j API (org.apache.logging.log4j:log4j-slf4j-impl:2.24.1 - https://logging.apache.org/log4j/2.x/log4j/log4j-slf4j-impl/) * SnakeYAML (org.yaml:snakeyaml:2.2 - https://bitbucket.org/snakeyaml/snakeyaml) @@ -374,7 +355,6 @@ List of third-party dependencies grouped by their license type. * Spring Messaging (org.springframework:spring-messaging:5.3.27 - https://github.com/spring-projects/spring-framework) * Spring Transaction (org.springframework:spring-tx:5.3.27 - https://github.com/spring-projects/spring-framework) * StAX API (stax:stax-api:1.0.1 - http://stax.codehaus.org/) - * T-Digest (com.tdunning:t-digest:3.2 - https://github.com/tdunning/t-digest) * Tephra API (co.cask.tephra:tephra-api:0.6.0 - https://github.com/caskdata/tephra/tephra-api) * Tephra Core (co.cask.tephra:tephra-core:0.6.0 - https://github.com/caskdata/tephra/tephra-core) * Tephra HBase 1.0 Compatibility (co.cask.tephra:tephra-hbase-compat-1.0:0.6.0 - https://github.com/caskdata/tephra/tephra-hbase-compat-1.0) @@ -426,10 +406,6 @@ List of third-party dependencies grouped by their license type. * Javassist (org.javassist:javassist:3.30.2-GA - https://www.javassist.org/) - Apache License, Version 2.0, LGPL-2.1-or-later - - * Java Native Access (net.java.dev.jna:jna:5.10.0 - https://github.com/java-native-access/jna) - Apache License (v2.0) * FlatBuffers Java API (com.vlkan:flatbuffers:1.2.0-3f79e055 - https://github.com/vy/flatbuffers) @@ -564,26 +540,6 @@ List of third-party dependencies grouped by their license type. * Jakarta Expression Language Implementation (org.glassfish:jakarta.el:4.0.2 - https://projects.eclipse.org/projects/ee4j.el) * jms (jakarta.jms:jakarta.jms-api:2.0.2 - https://projects.eclipse.org/projects/ee4j.jms) - Elastic License 2.0 - - * rest-high-level (org.elasticsearch.client:elasticsearch-rest-high-level-client:7.17.13 - https://github.com/elastic/elasticsearch.git) - - Elastic License 2.0, Server Side Public License, v 1 - - * aggs-matrix-stats (org.elasticsearch.plugin:aggs-matrix-stats-client:7.17.13 - https://github.com/elastic/elasticsearch.git) - * elasticsearch-cli (org.elasticsearch:elasticsearch-cli:7.17.13 - https://github.com/elastic/elasticsearch.git) - * elasticsearch-core (org.elasticsearch:elasticsearch-core:7.17.13 - https://github.com/elastic/elasticsearch.git) - * elasticsearch-geo (org.elasticsearch:elasticsearch-geo:7.17.13 - https://github.com/elastic/elasticsearch.git) - * elasticsearch-lz4 (org.elasticsearch:elasticsearch-lz4:7.17.13 - https://github.com/elastic/elasticsearch.git) - * elasticsearch-plugin-classloader (org.elasticsearch:elasticsearch-plugin-classloader:7.17.13 - https://github.com/elastic/elasticsearch.git) - * elasticsearch-secure-sm (org.elasticsearch:elasticsearch-secure-sm:7.17.13 - https://github.com/elastic/elasticsearch.git) - * elasticsearch-x-content (org.elasticsearch:elasticsearch-x-content:7.17.13 - https://github.com/elastic/elasticsearch.git) - * lang-mustache (org.elasticsearch.plugin:lang-mustache-client:7.17.13 - https://github.com/elastic/elasticsearch.git) - * mapper-extras (org.elasticsearch.plugin:mapper-extras-client:7.17.13 - https://github.com/elastic/elasticsearch.git) - * parent-join (org.elasticsearch.plugin:parent-join-client:7.17.13 - https://github.com/elastic/elasticsearch.git) - * rank-eval (org.elasticsearch.plugin:rank-eval-client:7.17.13 - https://github.com/elastic/elasticsearch.git) - * server (org.elasticsearch:elasticsearch:7.17.13 - https://github.com/elastic/elasticsearch.git) - MIT License * argparse4j (net.sourceforge.argparse4j:argparse4j:0.9.0 - https://argparse4j.github.io) @@ -591,7 +547,6 @@ List of third-party dependencies grouped by their license type. * JCodings (org.jruby.jcodings:jcodings:1.0.58 - http://nexus.sonatype.org/oss-repository-hosting.html/jcodings) * Jedis (redis.clients:jedis:5.1.0 - https://github.com/redis/jedis) * Joni (org.jruby.joni:joni:2.2.1 - http://nexus.sonatype.org/oss-repository-hosting.html/joni) - * JOpt Simple (net.sf.jopt-simple:jopt-simple:5.0.2 - http://pholser.github.io/jopt-simple) * JUL to SLF4J bridge (org.slf4j:jul-to-slf4j:1.7.36 - http://www.slf4j.org) * Microsoft JDBC Driver for SQL Server (com.microsoft.sqlserver:mssql-jdbc:6.2.1.jre7 - https://github.com/Microsoft/mssql-jdbc) * SLF4J API Module (org.slf4j:slf4j-api:1.7.36 - http://www.slf4j.org) diff --git a/docs/index.md b/docs/index.md index 36cf63f3ddd..7cdc90789d4 100644 --- a/docs/index.md +++ b/docs/index.md @@ -96,20 +96,9 @@ But small change will not affect the user experience. We will notify the user wh * [Apache HBase Integration](storm-hbase.html) * [Apache HDFS Integration](storm-hdfs.html) * [Apache Hive Integration](storm-hive.html) -* [Apache Solr Integration](storm-solr.html) -* [Apache Cassandra Integration](storm-cassandra.html) -* [Apache RocketMQ Integration](storm-rocketmq.html) * [JDBC Integration](storm-jdbc.html) * [JMS Integration](storm-jms.html) -* [MQTT Integration](storm-mqtt.html) * [Redis Integration](storm-redis.html) -* [Event Hubs Intergration](storm-eventhubs.html) -* [Elasticsearch Integration](storm-elasticsearch.html) -* [Mongodb Integration](storm-mongodb.html) -* [OpenTSDB Integration](storm-opentsdb.html) -* [Kinesis Integration](storm-kinesis.html) -* [PMML Integration](storm-pmml.html) -* [Kestrel Integration](Kestrel-and-Storm.html) #### Container, Resource Management System Integration diff --git a/docs/storm-elasticsearch.md b/docs/storm-elasticsearch.md deleted file mode 100644 index fff0f5168f0..00000000000 --- a/docs/storm-elasticsearch.md +++ /dev/null @@ -1,105 +0,0 @@ ---- -title: Storm Elasticsearch Integration -layout: documentation -documentation: true ---- - -# Storm Elasticsearch Bolt & Trident State - - EsIndexBolt, EsPercolateBolt and EsState allows users to stream data from storm into Elasticsearch directly. - For detailed description, please refer to the following. - -## EsIndexBolt (org.apache.storm.elasticsearch.bolt.EsIndexBolt) - -EsIndexBolt streams tuples directly into Elasticsearch. Tuples are indexed in specified index & type combination. -Users should make sure that ```EsTupleMapper``` can extract "source", "index", "type", and "id" from input tuple. -"index" and "type" are used for identifying target index and type. -"source" is a document in JSON format string that will be indexed in Elasticsearch. - -```java -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}); -EsTupleMapper tupleMapper = new DefaultEsTupleMapper(); -EsIndexBolt indexBolt = new EsIndexBolt(esConfig, tupleMapper); -``` - -## EsPercolateBolt (org.apache.storm.elasticsearch.bolt.EsPercolateBolt) - -EsPercolateBolt streams tuples directly into Elasticsearch. Tuples are used to send percolate request to specified index & type combination. -User should make sure ```EsTupleMapper``` can extract "source", "index", "type" from input tuple. -"index" and "type" are used for identifying target index and type. -"source" is a document in JSON format string that will be sent in percolate request to Elasticsearch. - -```java -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}); -EsTupleMapper tupleMapper = new DefaultEsTupleMapper(); -EsPercolateBolt percolateBolt = new EsPercolateBolt(esConfig, tupleMapper); -``` - -If there exists non-empty percolate response, EsPercolateBolt will emit tuple with original source and Percolate.Match -for each Percolate.Match in PercolateResponse. - -## EsState (org.apache.storm.elasticsearch.trident.EsState) - -Elasticsearch Trident state also follows similar pattern to EsBolts. It takes in EsConfig and EsTupleMapper as an arg. - -```java -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}); -EsTupleMapper tupleMapper = new DefaultEsTupleMapper(); - -StateFactory factory = new EsStateFactory(esConfig, tupleMapper); -TridentState state = stream.partitionPersist(factory, esFields, new EsUpdater(), new Fields()); - ``` - -## EsLookupBolt (org.apache.storm.elasticsearch.bolt.EsLookupBolt) - -EsLookupBolt performs a get request to Elasticsearch. -In order to do that, three dependencies need to be satisfied. Apart from usual EsConfig, two other dependencies must be provided: - ElasticsearchGetRequest is used to convert the incoming Tuple to the GetRequest that will be executed against Elasticsearch. - EsLookupResultOutput is used to declare the output fields and convert the GetResponse to values that are emited by the bolt. - -Incoming tuple is passed to provided GetRequest creator and the result of that execution is passed to Elasticsearch client. -The bolt then uses the provider output adapter (EsLookupResultOutput) to convert the GetResponse to Values to emit. -The output fields are also specified by the user of the bolt via the output adapter (EsLookupResultOutput). - -```java -EsConfig esConfig = createEsConfig(); -ElasticsearchGetRequest getRequestAdapter = createElasticsearchGetRequest(); -EsLookupResultOutput output = createOutput(); -EsLookupBolt lookupBolt = new EsLookupBolt(esConfig, getRequestAdapter, output); -``` - -## EsConfig (org.apache.storm.elasticsearch.common.EsConfig) - -Provided components (Bolt, State) takes in EsConfig as a constructor arg. - -```java -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}); -``` - -or - -```java -Map additionalParameters = new HashMap<>(); -additionalParameters.put("client.transport.sniff", "true"); -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}, additionalParameters); -``` - -### EsConfig params - -|Arg |Description | Type -|--- |--- |--- -|clusterName | Elasticsearch cluster name | String (required) | -|nodes | Elasticsearch nodes in a String array, each element should follow {host}:{port} pattern | String array (required) | -|additionalParameters | Additional Elasticsearch Transport Client configuration parameters | Map (optional) | - -## EsTupleMapper (org.apache.storm.elasticsearch.common.EsTupleMapper) - -For storing tuple to Elasticsearch or percolating tuple from Elasticsearch, we need to define which fields are used for. -Users need to define your own by implementing ```EsTupleMapper```. -Storm-elasticsearch presents default mapper ```org.apache.storm.elasticsearch.common.DefaultEsTupleMapper```, which extracts its source, index, type, id values from identical fields. -You can refer implementation of DefaultEsTupleMapper to see how to implement your own. - -## Committer Sponsors - - * Sriharsha Chintalapani ([@harshach](https://github.com/harshach)) - * Jungtaek Lim ([@HeartSaVioR](https://github.com/HeartSaVioR)) diff --git a/examples/storm-elasticsearch-examples/pom.xml b/examples/storm-elasticsearch-examples/pom.xml deleted file mode 100644 index b59c60a8c67..00000000000 --- a/examples/storm-elasticsearch-examples/pom.xml +++ /dev/null @@ -1,96 +0,0 @@ - - - - 4.0.0 - - - storm - org.apache.storm - 2.7.1-SNAPSHOT - ../../pom.xml - - - storm-elasticsearch-examples - - - - org.apache.storm - storm-client - ${project.version} - ${provided.scope} - - - org.apache.storm - storm-elasticsearch - ${project.version} - - - - - - - org.apache.maven.plugins - maven-shade-plugin - - true - - - *:* - - META-INF/*.SF - META-INF/*.sf - META-INF/*.DSA - META-INF/*.dsa - META-INF/*.RSA - META-INF/*.rsa - META-INF/*.EC - META-INF/*.ec - META-INF/MSFTSIG.SF - META-INF/MSFTSIG.RSA - - - - - - - package - - shade - - - - - - - - - - - - - org.apache.maven.plugins - maven-checkstyle-plugin - - - - org.apache.maven.plugins - maven-pmd-plugin - - - - diff --git a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/bolt/EsIndexTopology.java b/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/bolt/EsIndexTopology.java deleted file mode 100644 index c8e9c8948e4..00000000000 --- a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/bolt/EsIndexTopology.java +++ /dev/null @@ -1,187 +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.storm.elasticsearch.bolt; - -import java.util.Map; -import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; - -import org.apache.storm.Config; -import org.apache.storm.StormSubmitter; -import org.apache.storm.elasticsearch.common.DefaultEsTupleMapper; -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTupleMapper; -import org.apache.storm.generated.AlreadyAliveException; -import org.apache.storm.generated.AuthorizationException; -import org.apache.storm.generated.InvalidTopologyException; -import org.apache.storm.spout.SpoutOutputCollector; -import org.apache.storm.task.TopologyContext; -import org.apache.storm.topology.OutputFieldsDeclarer; -import org.apache.storm.topology.TopologyBuilder; -import org.apache.storm.topology.base.BaseRichSpout; -import org.apache.storm.tuple.Fields; -import org.apache.storm.tuple.Values; - -/** - * Demonstrates an ElasticSearch Storm topology. - */ -public final class EsIndexTopology { - - /** - * The id of the used spout. - */ - private static final String SPOUT_ID = "spout"; - /** - * The id of the used bolt. - */ - private static final String BOLT_ID = "bolt"; - /** - * The name of the used topology. - */ - private static final String TOPOLOGY_NAME = "elasticsearch-test-topology1"; - /** - * The number of pending tuples triggering logging. - */ - private static final int PENDING_COUNT_MAX = 1000; - - /** - * The example's main method. - * @param args the command line arguments - * @throws AlreadyAliveException if the topology is already started - * @throws InvalidTopologyException if the topology is invalid - * @throws AuthorizationException if the topology authorization fails - */ - public static void main(final String[] args) throws AlreadyAliveException, - InvalidTopologyException, - AuthorizationException { - Config config = new Config(); - config.setNumWorkers(1); - TopologyBuilder builder = new TopologyBuilder(); - UserDataSpout spout = new UserDataSpout(); - builder.setSpout(SPOUT_ID, spout, 1); - EsTupleMapper tupleMapper = new DefaultEsTupleMapper(); - EsConfig esConfig = new EsConfig("http://localhost:9300"); - builder.setBolt(BOLT_ID, new EsIndexBolt(esConfig, tupleMapper), 1) - .shuffleGrouping(SPOUT_ID); - - StormSubmitter.submitTopology(TOPOLOGY_NAME, - config, - builder.createTopology()); - } - - /** - * The user data spout. - */ - public static class UserDataSpout extends BaseRichSpout { - private static final long serialVersionUID = 1L; - /** - * The pending values. - */ - private ConcurrentHashMap pending; - /** - * The collector passed in - * {@link #open(java.util.Map, org.apache.storm.task.TopologyContext, - * org.apache.storm.spout.SpoutOutputCollector) }. - */ - private SpoutOutputCollector collector; - private String[] sources = { - "{\"user\":\"user1\"}", - "{\"user\":\"user2\"}", - "{\"user\":\"user3\"}", - "{\"user\":\"user4\"}" - }; - private int index = 0; - private int count = 0; - private long total = 0L; - private String indexName = "index1"; - private String typeName = "type1"; - - /** - * Declares {@code source}, {@code index}, {@code type} and {@code id}. - * @param declarer the declarer to pass to - */ - @Override - public void declareOutputFields(final OutputFieldsDeclarer declarer) { - declarer.declare(new Fields("source", "index", "type", "id")); - } - - /** - * Acquires {@code collector} and initializes {@code pending}. - * @param config unused - * @param context unused - * @param collectorArg the collector to acquire - */ - @Override - public void open(final Map config, - final TopologyContext context, - final SpoutOutputCollector collectorArg) { - this.collector = collectorArg; - this.pending = new ConcurrentHashMap<>(); - } - - /** - * Makes the spout emit the next tuple, if any. - */ - @Override - public void nextTuple() { - String source = sources[index]; - UUID msgId = UUID.randomUUID(); - Values values = new Values(source, indexName, typeName, msgId); - this.pending.put(msgId, values); - this.collector.emit(values, msgId); - index++; - if (index >= sources.length) { - index = 0; - } - count++; - total++; - if (count > PENDING_COUNT_MAX) { - count = 0; - System.out.println("Pending count: " + this.pending.size() - + ", total: " + this.total); - } - Thread.yield(); - } - - /** - * Acknowledges the message with id {@code msgId}. - * @param msgId the message id - */ - @Override - public void ack(final Object msgId) { - this.pending.remove(msgId); - } - - /** - * Marks the message with id {@code msgId} as failed. - * @param msgId the message id - */ - @Override - public void fail(final Object msgId) { - System.out.println("**** RESENDING FAILED TUPLE"); - this.collector.emit(this.pending.get(msgId), msgId); - } - } - - /** - * Utility constructor to prevent initialization. - */ - private EsIndexTopology() { - } -} diff --git a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/bolt/package-info.java b/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/bolt/package-info.java deleted file mode 100644 index 19fcc6a7c6d..00000000000 --- a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/bolt/package-info.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * Copyright 2018 The Apache Software Foundation. - * - * Licensed 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. - */ - -/** - * ElasticSearch examples classes. - */ -package org.apache.storm.elasticsearch.bolt; diff --git a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/common/EsConstants.java b/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/common/EsConstants.java deleted file mode 100644 index fd513d1443d..00000000000 --- a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/common/EsConstants.java +++ /dev/null @@ -1,40 +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.storm.elasticsearch.common; - -/** - * Constants used in ElasticSearch examples. - */ -public final class EsConstants { - - /** - * The cluster name. - */ - public static final String CLUSTER_NAME = "test-cluster"; - /** - * The default wait value in seconds. - */ - public static final int WAIT_DEFAULT_SECS = 5; - - /** - * Utility constructor. - */ - private EsConstants() { - } -} diff --git a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/common/package-info.java b/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/common/package-info.java deleted file mode 100644 index 9399cd2c5e7..00000000000 --- a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/common/package-info.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * Copyright 2018 The Apache Software Foundation. - * - * Licensed 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. - */ - -/** - * Shared code for example classes. - */ -package org.apache.storm.elasticsearch.common; diff --git a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/trident/TridentEsTopology.java b/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/trident/TridentEsTopology.java deleted file mode 100644 index d1a059f3c11..00000000000 --- a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/trident/TridentEsTopology.java +++ /dev/null @@ -1,210 +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.storm.elasticsearch.trident; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.UUID; - -import org.apache.storm.Config; -import org.apache.storm.StormSubmitter; -import org.apache.storm.elasticsearch.common.DefaultEsTupleMapper; -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTupleMapper; -import org.apache.storm.generated.AlreadyAliveException; -import org.apache.storm.generated.AuthorizationException; -import org.apache.storm.generated.InvalidTopologyException; -import org.apache.storm.task.TopologyContext; -import org.apache.storm.trident.Stream; -import org.apache.storm.trident.TridentState; -import org.apache.storm.trident.TridentTopology; -import org.apache.storm.trident.operation.TridentCollector; -import org.apache.storm.trident.spout.IBatchSpout; -import org.apache.storm.trident.state.StateFactory; -import org.apache.storm.tuple.Fields; -import org.apache.storm.tuple.Values; - -/** - * A Trident topology example. - */ -public final class TridentEsTopology { - private static final int BATCH_SIZE_DEFAULT = 100; - private static final String TOPOLOGY_NAME = "elasticsearch-test-topology2"; - - /** - * The example's main method. - * @param args the command line arguments - * @throws AlreadyAliveException if the topology is already started - * @throws InvalidTopologyException if the topology is invalid - * @throws AuthorizationException if the topology authorization fails - */ - public static void main(final String[] args) throws AlreadyAliveException, - InvalidTopologyException, - AuthorizationException { - int batchSize = BATCH_SIZE_DEFAULT; - FixedBatchSpout spout = new FixedBatchSpout(batchSize); - spout.cycle = true; - - TridentTopology topology = new TridentTopology(); - Stream stream = topology.newStream("spout", spout); - EsConfig esConfig = new EsConfig("http://localhost:9300"); - Fields esFields = new Fields("index", "type", "source"); - EsTupleMapper tupleMapper = new DefaultEsTupleMapper(); - StateFactory factory = new EsStateFactory(esConfig, tupleMapper); - TridentState state = stream.partitionPersist(factory, - esFields, - new EsUpdater(), - new Fields()); - - StormSubmitter.submitTopology(TOPOLOGY_NAME, - new Config(), - topology.build()); - } - - /** - * A fixed batch spout. - */ - public static class FixedBatchSpout implements IBatchSpout { - private static final long serialVersionUID = 1L; - private int maxBatchSize; - /** - * The passed batches. - */ - private HashMap>> batches = new HashMap<>(); - /** - * The output values. - */ - private Values[] outputs = { - new Values("{\"user\":\"user1\"}", - "index1", - "type1", - UUID.randomUUID().toString()), - new Values("{\"user\":\"user2\"}", - "index1", - "type2", - UUID.randomUUID().toString()), - new Values("{\"user\":\"user3\"}", - "index2", - "type1", - UUID.randomUUID().toString()), - new Values("{\"user\":\"user4\"}", - "index2", - "type2", - UUID.randomUUID().toString()) - }; - /** - * The current index. - */ - private int index = 0; - /** - * A flag indicating whether cycling ought to be performed. - */ - private boolean cycle = false; - - /** - * Creates a new fixed batch spout. - * @param maxBatchSizeArg the maximum batch size to set - */ - public FixedBatchSpout(final int maxBatchSizeArg) { - this.maxBatchSize = maxBatchSizeArg; - } - - /** - * Gets the output fields. - * @return the output fields. - */ - @Override - public Fields getOutputFields() { - return new Fields("source", "index", "type", "id"); - } - - /** - * Opens the spout. - * @param conf the configuration to use for opening - * @param context the context to use for opening - */ - @Override - public void open(final Map conf, - final TopologyContext context) { - index = 0; - } - - /** - * Emits a batch. - * @param batchId the batch id to use - * @param collector the collector to emit to - */ - @Override - public void emitBatch(final long batchId, - final TridentCollector collector) { - List> batch = this.batches.get(batchId); - if (batch == null) { - batch = new ArrayList>(); - if (index >= outputs.length && cycle) { - index = 0; - } - for (int i = 0; i < maxBatchSize; index++, i++) { - if (index == outputs.length) { - index = 0; - } - batch.add(outputs[index]); - } - this.batches.put(batchId, batch); - } - for (List list : batch) { - collector.emit(list); - } - } - - /** - * Acknowledges the message with id {@code msgId}. - * @param batchId the message id - */ - @Override - public void ack(final long batchId) { - this.batches.remove(batchId); - } - - /** - * Closes the spout. - */ - @Override - public void close() { - } - - /** - * Get the component configuration. - * @return the component configuration - */ - @Override - public Map getComponentConfiguration() { - Config conf = new Config(); - conf.setMaxTaskParallelism(1); - return conf; - } - } - - /** - * Utility constructor to prevent initialization. - */ - private TridentEsTopology() { - } -} diff --git a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/trident/package-info.java b/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/trident/package-info.java deleted file mode 100644 index 19369c31907..00000000000 --- a/examples/storm-elasticsearch-examples/src/main/java/org/apache/storm/elasticsearch/trident/package-info.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * Copyright 2018 The Apache Software Foundation. - * - * Licensed 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. - */ - -/** - * ElasticSearch Trident example classes. - */ -package org.apache.storm.elasticsearch.trident; diff --git a/external/storm-elasticsearch/README.md b/external/storm-elasticsearch/README.md deleted file mode 100644 index 2c52531c4eb..00000000000 --- a/external/storm-elasticsearch/README.md +++ /dev/null @@ -1,99 +0,0 @@ -# Storm Elasticsearch Bolt & Trident State - - EsIndexBolt, EsPercolateBolt and EsState allows users to stream data from storm into Elasticsearch directly. - For detailed description, please refer to the following. - -## EsIndexBolt (org.apache.storm.elasticsearch.bolt.EsIndexBolt) - -EsIndexBolt streams tuples directly into Elasticsearch. Tuples are indexed in specified index & type combination. -Users should make sure that ```EsTupleMapper``` can extract "source", "index", "type", and "id" from input tuple. -"index" and "type" are used for identifying target index and type. -"source" is a document in JSON format string that will be indexed in Elasticsearch. - -```java -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}); -EsTupleMapper tupleMapper = new DefaultEsTupleMapper(); -EsIndexBolt indexBolt = new EsIndexBolt(esConfig, tupleMapper); -``` - -## EsPercolateBolt (org.apache.storm.elasticsearch.bolt.EsPercolateBolt) - -EsPercolateBolt streams tuples directly into Elasticsearch. Tuples are used to send percolate request to specified index & type combination. -User should make sure ```EsTupleMapper``` can extract "source", "index", "type" from input tuple. -"index" and "type" are used for identifying target index and type. -"source" is a document in JSON format string that will be sent in percolate request to Elasticsearch. - -```java -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}); -EsTupleMapper tupleMapper = new DefaultEsTupleMapper(); -EsPercolateBolt percolateBolt = new EsPercolateBolt(esConfig, tupleMapper); -``` - -If there exists non-empty percolate response, EsPercolateBolt will emit tuple with original source and Percolate.Match -for each Percolate.Match in PercolateResponse. - -## EsState (org.apache.storm.elasticsearch.trident.EsState) - -Elasticsearch Trident state also follows similar pattern to EsBolts. It takes in EsConfig and EsTupleMapper as an arg. - -```java -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}); -EsTupleMapper tupleMapper = new DefaultEsTupleMapper(); - -StateFactory factory = new EsStateFactory(esConfig, tupleMapper); -TridentState state = stream.partitionPersist(factory, esFields, new EsUpdater(), new Fields()); - ``` - -## EsLookupBolt (org.apache.storm.elasticsearch.bolt.EsLookupBolt) - -EsLookupBolt performs a get request to Elasticsearch. -In order to do that, three dependencies need to be satisfied. Apart from usual EsConfig, two other dependencies must be provided: - ElasticsearchGetRequest is used to convert the incoming Tuple to the GetRequest that will be executed against Elasticsearch. - EsLookupResultOutput is used to declare the output fields and convert the GetResponse to values that are emited by the bolt. - -Incoming tuple is passed to provided GetRequest creator and the result of that execution is passed to Elasticsearch client. -The bolt then uses the provider output adapter (EsLookupResultOutput) to convert the GetResponse to Values to emit. -The output fields are also specified by the user of the bolt via the output adapter (EsLookupResultOutput). - -```java -EsConfig esConfig = createEsConfig(); -ElasticsearchGetRequest getRequestAdapter = createElasticsearchGetRequest(); -EsLookupResultOutput output = createOutput(); -EsLookupBolt lookupBolt = new EsLookupBolt(esConfig, getRequestAdapter, output); -``` - -## EsConfig (org.apache.storm.elasticsearch.common.EsConfig) - -Provided components (Bolt, State) takes in EsConfig as a constructor arg. - -```java -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}); -``` - -or - -```java -Map additionalParameters = new HashMap<>(); -additionalParameters.put("client.transport.sniff", "true"); -EsConfig esConfig = new EsConfig(clusterName, new String[]{"localhost:9300"}, additionalParameters); -``` - -### EsConfig params - -|Arg |Description | Type -|--- |--- |--- -|clusterName | Elasticsearch cluster name | String (required) | -|nodes | Elasticsearch nodes in a String array, each element should follow {host}:{port} pattern | String array (required) | -|additionalParameters | Additional Elasticsearch Transport Client configuration parameters | Map (optional) | - -## EsTupleMapper (org.apache.storm.elasticsearch.common.EsTupleMapper) - -For storing tuple to Elasticsearch or percolating tuple from Elasticsearch, we need to define which fields are used for. -Users need to define your own by implementing ```EsTupleMapper```. -Storm-elasticsearch presents default mapper ```org.apache.storm.elasticsearch.common.DefaultEsTupleMapper```, which extracts its source, index, type, id values from identical fields. -You can refer implementation of DefaultEsTupleMapper to see how to implement your own. - -## Committer Sponsors - - * Sriharsha Chintalapani ([@harshach](https://github.com/harshach)) - * Jungtaek Lim ([@HeartSaVioR](https://github.com/HeartSaVioR)) diff --git a/external/storm-elasticsearch/pom.xml b/external/storm-elasticsearch/pom.xml deleted file mode 100644 index b4f3410cdbf..00000000000 --- a/external/storm-elasticsearch/pom.xml +++ /dev/null @@ -1,188 +0,0 @@ - - - - 4.0.0 - - - storm - org.apache.storm - 2.7.1-SNAPSHOT - ../../pom.xml - - - storm-elasticsearch - - - - sweetest - Adrian Seungjin Lee - sweetest.sj@navercorp.com - - - hmcc - Heather McCartney - heather@problemchimp.org - - - - - 2.4.4 - - - - - commons-io - commons-io - compile - - - org.apache.storm - storm-client - ${project.version} - ${provided.scope} - - - org.elasticsearch.client - elasticsearch-rest-high-level-client - ${elasticsearch.version} - - - org.apache.httpcomponents - httpasyncclient - 4.1.5 - - - org.apache.httpcomponents - httpclient - - - org.apache.httpcomponents - httpcore - - - - - org.apache.httpcomponents - httpclient - - - org.apache.httpcomponents - httpcore - - - com.fasterxml.jackson.core - jackson-core - - - com.fasterxml.jackson.core - jackson-databind - - - com.google.guava - guava-testlib - ${guava.version} - test - - - org.mockito - mockito-core - - - org.mockito - mockito-junit-jupiter - - - org.hamcrest - hamcrest - - - org.apache.storm - storm-client - ${project.version} - test-jar - test - - - com.google.guava - guava - test - - - org.testcontainers - elasticsearch - ${testcontainers.version} - test - - - - - - - maven-clean-plugin - - - cleanup - test-compile - - clean - - - true - - - ./data/ - - - - - - - - maven-surefire-plugin - - - default-test - test - - test - - - - - - ${elasticsearch.version} - - - - - org.apache.maven.plugins - maven-failsafe-plugin - - - org.apache.maven.plugins - maven-checkstyle-plugin - - - - org.apache.maven.plugins - maven-pmd-plugin - - - - - diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/DefaultEsLookupResultOutput.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/DefaultEsLookupResultOutput.java deleted file mode 100644 index d25beadb5b7..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/DefaultEsLookupResultOutput.java +++ /dev/null @@ -1,67 +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.storm.elasticsearch; - -import com.fasterxml.jackson.databind.ObjectMapper; - -import java.io.IOException; -import java.util.Collection; -import java.util.Collections; - -import org.apache.storm.elasticsearch.response.LookupResponse; -import org.apache.storm.tuple.Fields; -import org.apache.storm.tuple.Values; -import org.elasticsearch.client.Response; - -/** - * Default implementation of {@link EsLookupResultOutput}. - * Outputs the index, type, id and source as strings. - */ -public class DefaultEsLookupResultOutput implements EsLookupResultOutput { - - private static final long serialVersionUID = 2932278450655703239L; - - private ObjectMapper objectMapper; - - public DefaultEsLookupResultOutput(ObjectMapper objectMapper) { - super(); - this.objectMapper = objectMapper; - } - - @Override - public Collection toValues(Response response) { - LookupResponse lookupResponse; - try { - lookupResponse = objectMapper.readValue(response.getEntity().getContent(), LookupResponse.class); - } catch (UnsupportedOperationException | IOException e) { - throw new IllegalArgumentException("Response " + response + " is invalid", e); - } - return Collections.singleton(new Values( - lookupResponse.getIndex(), - lookupResponse.getType(), - lookupResponse.getId(), - lookupResponse.getSource())); - } - - @Override - public Fields fields() { - return new Fields("index", "type", "id", "source"); - } - -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/EsLookupResultOutput.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/EsLookupResultOutput.java deleted file mode 100644 index 71e250d7fbd..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/EsLookupResultOutput.java +++ /dev/null @@ -1,48 +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.storm.elasticsearch; - -import java.io.Serializable; -import java.util.Collection; - -import org.apache.storm.tuple.Fields; -import org.apache.storm.tuple.Values; -import org.elasticsearch.client.Response; - -/** - * The adapter to convert the results fetched from Elasticsearch to values. - * - * @since 0.11 - */ -public interface EsLookupResultOutput extends Serializable { - - /** - * Convert Elasticsearch response to a collection of {@link Values}. - * - * @return collection of values to emit. - */ - Collection toValues(Response response); - - /** - * {@link Fields} to be declared for output. - * - * @return output fields to declare. - */ - Fields fields(); -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/AbstractEsBolt.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/AbstractEsBolt.java deleted file mode 100644 index 5612e707354..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/AbstractEsBolt.java +++ /dev/null @@ -1,100 +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.storm.elasticsearch.bolt; - -import static java.util.Objects.requireNonNull; -import static org.apache.http.util.Args.notBlank; - -import com.fasterxml.jackson.databind.ObjectMapper; - -import java.util.Map; - -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.StormElasticSearchClient; -import org.apache.storm.task.OutputCollector; -import org.apache.storm.task.TopologyContext; -import org.apache.storm.topology.OutputFieldsDeclarer; -import org.apache.storm.topology.base.BaseTickTupleAwareRichBolt; -import org.elasticsearch.client.RestClient; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public abstract class AbstractEsBolt extends BaseTickTupleAwareRichBolt { - - private static final Logger LOG = LoggerFactory.getLogger(AbstractEsBolt.class); - - protected static RestClient client; - protected static final ObjectMapper objectMapper = new ObjectMapper(); - - protected OutputCollector collector; - private EsConfig esConfig; - - public AbstractEsBolt(EsConfig esConfig) { - this.esConfig = requireNonNull(esConfig); - } - - @Override - public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) { - try { - this.collector = outputCollector; - synchronized (AbstractEsBolt.class) { - if (client == null) { - client = new StormElasticSearchClient(esConfig).construct(); - } - } - } catch (Exception e) { - LOG.warn("unable to initialize EsBolt ", e); - } - } - - @Override - public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { - } - - /** - * Construct an Elasticsearch endpoint from the provided index, type and - * id. - * @param index - required; name of Elasticsearch index - * @param type - optional; name of Elasticsearch type - * @param id - optional; Elasticsearch document ID - * @return the index, type and id concatenated with '/'. - */ - static String getEndpoint(String index, String type, String id) { - requireNonNull(index); - notBlank(index, "index"); - - StringBuilder sb = new StringBuilder(); - sb.append("/").append(index); - if (!(type == null || type.isEmpty())) { - sb.append("/").append(type); - } - if (!(id == null || id.isEmpty())) { - sb.append("/").append(id); - } - return sb.toString(); - } - - static RestClient getClient() { - return AbstractEsBolt.client; - } - - static void replaceClient(RestClient client) { - AbstractEsBolt.client = client; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsIndexBolt.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsIndexBolt.java deleted file mode 100644 index 9d60b978607..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsIndexBolt.java +++ /dev/null @@ -1,94 +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.storm.elasticsearch.bolt; - -import static java.util.Objects.requireNonNull; - -import java.util.HashMap; -import java.util.Map; - -import org.apache.http.entity.StringEntity; -import org.apache.storm.elasticsearch.common.DefaultEsTupleMapper; -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTupleMapper; -import org.apache.storm.task.OutputCollector; -import org.apache.storm.task.TopologyContext; -import org.apache.storm.topology.OutputFieldsDeclarer; -import org.apache.storm.tuple.Tuple; -import org.elasticsearch.client.Request; -import org.elasticsearch.client.Response; - -/** - * Basic bolt for storing tuple to ES document. - */ -public class EsIndexBolt extends AbstractEsBolt { - private final EsTupleMapper tupleMapper; - - /** - * EsIndexBolt constructor. - * @param esConfig Elasticsearch configuration containing node addresses {@link EsConfig} - */ - public EsIndexBolt(EsConfig esConfig) { - this(esConfig, new DefaultEsTupleMapper()); - } - - /** - * EsIndexBolt constructor. - * @param esConfig Elasticsearch configuration containing node addresses {@link EsConfig} - * @param tupleMapper Tuple to ES document mapper {@link EsTupleMapper} - */ - public EsIndexBolt(EsConfig esConfig, EsTupleMapper tupleMapper) { - super(esConfig); - this.tupleMapper = requireNonNull(tupleMapper); - } - - @Override - public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) { - super.prepare(map, topologyContext, outputCollector); - } - - /** - * {@inheritDoc} - * Tuple should have relevant fields (source, index, type, id) for tupleMapper to extract ES document. - */ - @Override - public void process(Tuple tuple) { - try { - String source = tupleMapper.getSource(tuple); - String index = tupleMapper.getIndex(tuple); - String type = tupleMapper.getType(tuple); - String id = tupleMapper.getId(tuple); - Map params = tupleMapper.getParams(tuple, new HashMap<>()); - - final Request request = new Request("post", getEndpoint(index, type, id)); - request.setEntity(new StringEntity(source)); - request.addParameters(params); - - client.performRequest(request); - collector.ack(tuple); - } catch (Exception e) { - collector.reportError(e); - collector.fail(tuple); - } - } - - @Override - public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupBolt.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupBolt.java deleted file mode 100644 index f84b5b601fd..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupBolt.java +++ /dev/null @@ -1,105 +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.storm.elasticsearch.bolt; - -import static java.util.Objects.requireNonNull; - -import java.io.IOException; -import java.util.Collection; -import java.util.HashMap; -import java.util.Map; - -import org.apache.http.entity.StringEntity; -import org.apache.storm.elasticsearch.DefaultEsLookupResultOutput; -import org.apache.storm.elasticsearch.EsLookupResultOutput; -import org.apache.storm.elasticsearch.common.DefaultEsTupleMapper; -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTupleMapper; -import org.apache.storm.topology.OutputFieldsDeclarer; -import org.apache.storm.tuple.Tuple; -import org.apache.storm.tuple.Values; -import org.elasticsearch.client.Request; -import org.elasticsearch.client.Response; - -/** - * Basic bolt for looking up document in ES. - * @since 0.11 - */ -public class EsLookupBolt extends AbstractEsBolt { - - private final EsTupleMapper tupleMapper; - private final EsLookupResultOutput output; - - /** - * EsLookupBolt constructor. - * @param esConfig Elasticsearch configuration containing node addresses {@link EsConfig} - * @throws NullPointerException if any of the parameters is null - */ - public EsLookupBolt(EsConfig esConfig) { - this(esConfig, new DefaultEsTupleMapper(), new DefaultEsLookupResultOutput(objectMapper)); - } - - /** - * EsLookupBolt constructor. - * @param esConfig Elasticsearch configuration containing node addresses {@link EsConfig} - * @param tupleMapper Tuple to ES document mapper {@link EsTupleMapper} - * @param output ES response to Values mapper {@link EsLookupResultOutput} - * @throws NullPointerException if any of the parameters is null - */ - public EsLookupBolt(EsConfig esConfig, EsTupleMapper tupleMapper, EsLookupResultOutput output) { - super(esConfig); - this.tupleMapper = requireNonNull(tupleMapper); - this.output = requireNonNull(output); - } - - @Override - public void process(Tuple tuple) { - try { - Collection values = lookupValuesInEs(tuple); - tryEmitAndAck(values, tuple); - } catch (Exception e) { - collector.reportError(e); - collector.fail(tuple); - } - } - - private Collection lookupValuesInEs(Tuple tuple) throws IOException { - String index = tupleMapper.getIndex(tuple); - String type = tupleMapper.getType(tuple); - String id = tupleMapper.getId(tuple); - Map params = tupleMapper.getParams(tuple, new HashMap<>()); - - final Request request = new Request("get", getEndpoint(index, type, id)); - request.addParameters(params); - Response response = client.performRequest(request); - return output.toValues(response); - } - - private void tryEmitAndAck(Collection values, Tuple tuple) { - for (Values value : values) { - collector.emit(tuple, value); - } - collector.ack(tuple); - } - - @Override - public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { - outputFieldsDeclarer.declare(output.fields()); - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsPercolateBolt.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsPercolateBolt.java deleted file mode 100644 index 2fac3e8f689..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsPercolateBolt.java +++ /dev/null @@ -1,110 +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.storm.elasticsearch.bolt; - -import static java.util.Objects.requireNonNull; - -import java.util.HashMap; -import java.util.Map; - -import org.apache.http.entity.StringEntity; -import org.apache.storm.elasticsearch.common.DefaultEsTupleMapper; -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTupleMapper; -import org.apache.storm.elasticsearch.response.PercolateResponse; -import org.apache.storm.task.OutputCollector; -import org.apache.storm.task.TopologyContext; -import org.apache.storm.topology.OutputFieldsDeclarer; -import org.apache.storm.tuple.Fields; -import org.apache.storm.tuple.Tuple; -import org.apache.storm.tuple.Values; -import org.elasticsearch.client.Request; -import org.elasticsearch.client.Response; - -/** - * Basic bolt for retrieve matched percolate queries. - */ -public class EsPercolateBolt extends AbstractEsBolt { - - private final EsTupleMapper tupleMapper; - - /** - * EsPercolateBolt constructor. - * @param esConfig Elasticsearch configuration containing node addresses {@link EsConfig} - */ - public EsPercolateBolt(EsConfig esConfig) { - this(esConfig, new DefaultEsTupleMapper()); - } - - /** - * EsPercolateBolt constructor. - * @param esConfig Elasticsearch configuration containing node addresses and cluster name {@link EsConfig} - * @param tupleMapper Tuple to ES document mapper {@link EsTupleMapper} - */ - public EsPercolateBolt(EsConfig esConfig, EsTupleMapper tupleMapper) { - super(esConfig); - this.tupleMapper = requireNonNull(tupleMapper); - } - - @Override - public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) { - super.prepare(map, topologyContext, outputCollector); - } - - /** - * {@inheritDoc} - * Tuple should have relevant fields (source, index, type) for storeMapper to extract ES document.
- * If there exists non-empty percolate response, EsPercolateBolt will emit tuple with original source - * and Percolate.Match for each Percolate.Match in PercolateResponse. - */ - @Override - public void process(Tuple tuple) { - try { - String source = tupleMapper.getSource(tuple); - String index = tupleMapper.getIndex(tuple); - String type = tupleMapper.getType(tuple); - - Map indexParams = new HashMap<>(); - indexParams.put(type, null); - String percolateDoc = "{\"doc\": " + source + "}"; - - final Request request = new Request("get", getEndpoint(index, type, "_percolate")); - request.setEntity(new StringEntity(percolateDoc)); - request.addParameters(new HashMap<>()); - - Response response = client.performRequest(request); - - PercolateResponse percolateResponse = objectMapper.readValue(response.getEntity().getContent(), PercolateResponse.class); - if (!percolateResponse.getMatches().isEmpty()) { - for (PercolateResponse.Match match : percolateResponse.getMatches()) { - collector.emit(new Values(source, match)); - } - } - collector.ack(tuple); - } catch (Exception e) { - collector.reportError(e); - collector.fail(tuple); - } - } - - @Override - public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { - outputFieldsDeclarer.declare(new Fields("source", "match")); - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/DefaultEsTupleMapper.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/DefaultEsTupleMapper.java deleted file mode 100644 index af4f4e9804a..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/DefaultEsTupleMapper.java +++ /dev/null @@ -1,62 +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.storm.elasticsearch.common; - -import java.util.HashMap; -import java.util.Map; - -import org.apache.storm.tuple.ITuple; - -public class DefaultEsTupleMapper implements EsTupleMapper { - @Override - public String getSource(ITuple tuple) { - return tuple.getStringByField("source"); - } - - @Override - public String getIndex(ITuple tuple) { - return tuple.getStringByField("index"); - } - - @Override - public String getType(ITuple tuple) { - return tuple.getStringByField("type"); - } - - @Override - public String getId(ITuple tuple) { - return tuple.getStringByField("id"); - } - - @Override - public Map getParams(ITuple tuple, Map defaultValue) { - if (!tuple.contains("params")) { - return defaultValue; - } - Object o = tuple.getValueByField("params"); - if (o instanceof Map) { - Map params = new HashMap<>(); - for (Map.Entry entry : ((Map) o).entrySet()) { - params.put(entry.getKey().toString(), entry.getValue() == null ? null : entry.getValue().toString()); - } - return params; - } - return defaultValue; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/EsConfig.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/EsConfig.java deleted file mode 100644 index 04ac73d4cd2..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/EsConfig.java +++ /dev/null @@ -1,138 +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.storm.elasticsearch.common; - -import java.io.Serializable; -import java.net.URI; -import java.net.URISyntaxException; - -import org.apache.http.Header; -import org.apache.http.HttpHost; -import org.elasticsearch.client.RestClient; -import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback; -import org.elasticsearch.client.RestClientBuilder.RequestConfigCallback; - -/** - * Configuration for Elasticsearch connection. - * @since 0.11 - */ -public class EsConfig implements Serializable { - - private final HttpHost[] httpHosts; - private Header[] defaultHeaders; - private RestClient.FailureListener failureListener; - private HttpClientConfigCallback httpClientConfigCallback; - private RequestConfigCallback requestConfigCallback; - private String pathPrefix; - - private boolean compression; - - /** - * EsConfig Constructor to be used in EsIndexBolt, EsPercolateBolt and EsStateFactory. - * Connects to Elasticsearch at http://localhost:9200. - */ - public EsConfig() { - this("http://localhost:9200"); - } - - /** - * EsConfig Constructor to be used in EsIndexBolt, EsPercolateBolt and EsStateFactory. - * - * @param urls Elasticsearch addresses in scheme://host:port pattern string array - * @throws IllegalArgumentException if urls are empty - * @throws NullPointerException on any of the fields being null - */ - public EsConfig(String... urls) { - if (urls.length == 0) { - throw new IllegalArgumentException("urls is required"); - } - this.httpHosts = new HttpHost[urls.length]; - for (int i = 0; i < urls.length; i++) { - URI uri = toUri(urls[i]); - this.httpHosts[i] = new HttpHost(uri.getHost(), uri.getPort(), uri.getScheme()); - } - } - - static URI toUri(String url) throws IllegalArgumentException { - try { - return new URI(url); - } catch (URISyntaxException e) { - throw new IllegalArgumentException("Invalid url " + url); - } - } - - public EsConfig withDefaultHeaders(Header[] defaultHeaders) { - this.defaultHeaders = defaultHeaders; - return this; - } - - public EsConfig withFailureListener(RestClient.FailureListener failureListener) { - this.failureListener = failureListener; - return this; - } - - public EsConfig withHttpClientConfigCallback(HttpClientConfigCallback httpClientConfigCallback) { - this.httpClientConfigCallback = httpClientConfigCallback; - return this; - } - - public EsConfig withRequestConfigCallback(RequestConfigCallback requestConfigCallback) { - this.requestConfigCallback = requestConfigCallback; - return this; - } - - public EsConfig withPathPrefix(String pathPrefix) { - this.pathPrefix = pathPrefix; - return this; - } - - public EsConfig withCompression(boolean compression) { - this.compression = compression; - return this; - } - - public HttpHost[] getHttpHosts() { - return httpHosts; - } - - public Header[] getDefaultHeaders() { - return defaultHeaders; - } - - public RestClient.FailureListener getFailureListener() { - return failureListener; - } - - public HttpClientConfigCallback getHttpClientConfigCallback() { - return httpClientConfigCallback; - } - - public RequestConfigCallback getRequestConfigCallback() { - return requestConfigCallback; - } - - public String getPathPrefix() { - return pathPrefix; - } - - public boolean isCompression() { - return compression; - } - -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/EsTupleMapper.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/EsTupleMapper.java deleted file mode 100644 index ab658da1257..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/EsTupleMapper.java +++ /dev/null @@ -1,65 +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.storm.elasticsearch.common; - -import java.io.Serializable; -import java.util.Map; - -import org.apache.storm.tuple.ITuple; - -/** - * TupleMapper defines how to extract source, index, type, and id from tuple for ElasticSearch. - */ -public interface EsTupleMapper extends Serializable { - /** - * Extracts source from tuple. - * @param tuple source tuple - * @return source - */ - String getSource(ITuple tuple); - - /** - * Extracts index from tuple. - * @param tuple source tuple - * @return index - */ - String getIndex(ITuple tuple); - - /** - * Extracts type from tuple. - * @param tuple source tuple - * @return type - */ - String getType(ITuple tuple); - - /** - * Extracts id from tuple. - * @param tuple source tuple - * @return id - */ - String getId(ITuple tuple); - - /** - * Extracts params from tuple if available. - * @param tuple source tuple - * @param defaultValue value to return if params are missing - * @return params - */ - Map getParams(ITuple tuple, Map defaultValue); -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/StormElasticSearchClient.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/StormElasticSearchClient.java deleted file mode 100644 index 7fa57d7b956..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/StormElasticSearchClient.java +++ /dev/null @@ -1,64 +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.storm.elasticsearch.common; - -import java.io.Serializable; - -import org.elasticsearch.client.RestClient; -import org.elasticsearch.client.RestClientBuilder; - -/** - * Client for connecting Elasticsearch. - */ -public final class StormElasticSearchClient implements Serializable { - - private final EsConfig esConfig; - - public StormElasticSearchClient(EsConfig esConfig) { - this.esConfig = esConfig; - } - - /** - * Creates a new {@link RestClient} using given {@link EsConfig}. - * - * @return {@link RestClient} for Elasticsearch connection - */ - public RestClient construct() { - RestClientBuilder builder = RestClient.builder(esConfig.getHttpHosts()); - if (esConfig.getDefaultHeaders() != null) { - builder.setDefaultHeaders(esConfig.getDefaultHeaders()); - } - if (esConfig.getFailureListener() != null) { - builder.setFailureListener(esConfig.getFailureListener()); - } - if (esConfig.getHttpClientConfigCallback() != null) { - builder.setHttpClientConfigCallback(esConfig.getHttpClientConfigCallback()); - } - if (esConfig.getRequestConfigCallback() != null) { - builder.setRequestConfigCallback(esConfig.getRequestConfigCallback()); - } - if (esConfig.getPathPrefix() != null) { - builder.setPathPrefix(esConfig.getPathPrefix()); - } - - builder.setCompressionEnabled(esConfig.isCompression()); - - return builder.build(); - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/Index.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/Index.java deleted file mode 100644 index 463061fb599..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/Index.java +++ /dev/null @@ -1,77 +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.storm.elasticsearch.doc; - -import com.fasterxml.jackson.annotation.JsonProperty; - -/** - * Elasticsearch document fragment with "_index", "_type" and "_id" fields. - */ -public class Index { - - public Index() { - - } - - /** - * Create a Index with the specified index, type and id. - * - * @param index index name - * @param type document type to be stored - * @param id unique document id in Elasticsearch - */ - public Index(String index, String type, String id) { - this.index = index; - this.type = type; - this.id = id; - } - - @JsonProperty("_index") - private String index; - - @JsonProperty("_type") - private String type; - - @JsonProperty("_id") - private String id; - - public String getIndex() { - return index; - } - - public void setIndex(String index) { - this.index = index; - } - - public String getType() { - return type; - } - - public void setType(String type) { - this.type = type; - } - - public String getId() { - return id; - } - - public void setId(String id) { - this.id = id; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/IndexDoc.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/IndexDoc.java deleted file mode 100644 index 1ea2b4bb167..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/IndexDoc.java +++ /dev/null @@ -1,44 +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.storm.elasticsearch.doc; - -/** - * Elasticsearch document fragment with a single "index" field, used in bulk - * requests. - */ -public class IndexDoc { - - public IndexDoc() { - - } - - public IndexDoc(Index index) { - this.index = index; - } - - private Index index; - - public Index getIndex() { - return index; - } - - public void setIndex(Index index) { - this.index = index; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/IndexItem.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/IndexItem.java deleted file mode 100644 index c35dfe4bca1..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/IndexItem.java +++ /dev/null @@ -1,92 +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.storm.elasticsearch.doc; - -import com.fasterxml.jackson.annotation.JsonProperty; -import com.fasterxml.jackson.annotation.JsonRawValue; -import com.fasterxml.jackson.databind.JsonNode; - -/** - * Elasticsearch document fragment containing extended index information, used - * in bulk index responses. - */ -public class IndexItem extends Index { - - @JsonProperty("_version") - private long version; - - private String result; - private boolean created; - private int status; - - @JsonRawValue - private Object error; - - - @JsonProperty("_shards") - private Shards shards; - - public long getVersion() { - return version; - } - - public void setVersion(long version) { - this.version = version; - } - - public String getResult() { - return result; - } - - public void setResult(String result) { - this.result = result; - } - - public boolean isCreated() { - return created; - } - - public void setCreated(boolean created) { - this.created = created; - } - - public int getStatus() { - return status; - } - - public void setStatus(int status) { - this.status = status; - } - - public String getError() { - return error == null ? null : error.toString(); - } - - public void setError(JsonNode error) { - this.error = error; - } - - public Shards getShards() { - return shards; - } - - public void setShards(Shards shards) { - this.shards = shards; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/IndexItemDoc.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/IndexItemDoc.java deleted file mode 100644 index a7bf789e779..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/IndexItemDoc.java +++ /dev/null @@ -1,43 +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.storm.elasticsearch.doc; - -/** - * Elasticsearch document with a single "index" field, used in bulk responses. - */ -public class IndexItemDoc { - - public IndexItemDoc() { - - } - - public IndexItemDoc(IndexItem index) { - this.index = index; - } - - private IndexItem index; - - public IndexItem getIndex() { - return index; - } - - public void setIndex(IndexItem index) { - this.index = index; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/Shards.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/Shards.java deleted file mode 100644 index 1bc7e66a1dd..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/Shards.java +++ /dev/null @@ -1,64 +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.storm.elasticsearch.doc; - -import java.util.List; - -/** - * Elasticsearch document fragment containing shard success information, - * used in percolate and bulk index responses. - */ -public class Shards { - private int total; - private int successful; - private int failed; - private List failures; - - public int getTotal() { - return total; - } - - public void setTotal(int total) { - this.total = total; - } - - public int getSuccessful() { - return successful; - } - - public void setSuccessful(int successful) { - this.successful = successful; - } - - public int getFailed() { - return failed; - } - - public void setFailed(int failed) { - this.failed = failed; - } - - public List getFailures() { - return failures; - } - - public void setFailures(List failures) { - this.failures = failures; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/SourceDoc.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/SourceDoc.java deleted file mode 100644 index 6813d4da33e..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/doc/SourceDoc.java +++ /dev/null @@ -1,44 +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.storm.elasticsearch.doc; - -/** - * Elasticsearch document fragment with a single field, "source", used in bulk - * requests. - */ -public class SourceDoc { - - public SourceDoc() { - - } - - public SourceDoc(String source) { - this.source = source; - } - - private String source; - - public String getSource() { - return source; - } - - public void setSource(String source) { - this.source = source; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/response/BulkIndexResponse.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/response/BulkIndexResponse.java deleted file mode 100644 index a0f40c87dba..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/response/BulkIndexResponse.java +++ /dev/null @@ -1,91 +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.storm.elasticsearch.response; - -import java.util.List; - -import org.apache.storm.elasticsearch.doc.IndexItemDoc; - -/** - * Mapped response for bulk index. - */ -public class BulkIndexResponse { - - private boolean errors; - private long took; - private List items; - - public BulkIndexResponse() { - - } - - public boolean hasErrors() { - return errors; - } - - public void setErrors(boolean errors) { - this.errors = errors; - } - - public long getTook() { - return took; - } - - public void setTook(long took) { - this.took = took; - } - - public List getItems() { - return items; - } - - public void setItems(List items) { - this.items = items; - } - - /** - * Retrieve first error's code from response. - * - * @return error status code - */ - public Integer getFirstError() { - if (items == null || items.isEmpty()) { - return null; - } - for (IndexItemDoc item : items) { - int status = item.getIndex().getStatus(); - if (400 <= status && status <= 599) { - return status; - } - } - return null; - } - - /** - * Retrieve first result from response. - * - * @return result text - */ - public String getFirstResult() { - if (items == null || items.isEmpty()) { - return null; - } - return items.get(0).getIndex().getResult(); - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/response/LookupResponse.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/response/LookupResponse.java deleted file mode 100644 index 99730f7ce7b..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/response/LookupResponse.java +++ /dev/null @@ -1,64 +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.storm.elasticsearch.response; - -import com.fasterxml.jackson.annotation.JsonProperty; -import com.fasterxml.jackson.annotation.JsonRawValue; -import com.fasterxml.jackson.databind.JsonNode; - -import org.apache.storm.elasticsearch.doc.Index; - -/** - * Mapped response for document lookup. - */ -public class LookupResponse extends Index { - - @JsonProperty("_version") - private long version; - - private boolean found; - - @JsonProperty("_source") - @JsonRawValue - private Object source; - - public long getVersion() { - return version; - } - - public void setVersion(long version) { - this.version = version; - } - - public boolean isFound() { - return found; - } - - public void setFound(boolean found) { - this.found = found; - } - - public String getSource() { - return source == null ? null : source.toString(); - } - - public void setSource(JsonNode source) { - this.source = source; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/response/PercolateResponse.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/response/PercolateResponse.java deleted file mode 100644 index df671344fa5..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/response/PercolateResponse.java +++ /dev/null @@ -1,86 +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.storm.elasticsearch.response; - -import com.fasterxml.jackson.annotation.JsonProperty; - -import java.util.List; - -import org.apache.storm.elasticsearch.doc.Shards; - -/** - * Mapped response for percolate. - */ -public class PercolateResponse { - - private long took; - private long total; - private List matches; - - @JsonProperty("_shards") - private Shards shards; - - public static class Match { - @JsonProperty("_index") - private String index; - - @JsonProperty("_id") - private String id; - - public String getIndex() { - return index; - } - - public void setIndex(String index) { - this.index = index; - } - - public String getId() { - return id; - } - - public void setId(String id) { - this.id = id; - } - } - - public long getTook() { - return took; - } - - public void setTook(long took) { - this.took = took; - } - - public long getTotal() { - return total; - } - - public void setTotal(long total) { - this.total = total; - } - - public List getMatches() { - return matches; - } - - public void setMatches(List matches) { - this.matches = matches; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsState.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsState.java deleted file mode 100644 index ef82335f83f..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsState.java +++ /dev/null @@ -1,139 +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.storm.elasticsearch.trident; - -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; - -import java.io.IOException; -import java.util.List; - -import org.apache.http.entity.StringEntity; -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTupleMapper; -import org.apache.storm.elasticsearch.common.StormElasticSearchClient; -import org.apache.storm.elasticsearch.doc.Index; -import org.apache.storm.elasticsearch.doc.IndexDoc; -import org.apache.storm.elasticsearch.doc.SourceDoc; -import org.apache.storm.elasticsearch.response.BulkIndexResponse; -import org.apache.storm.topology.FailedException; -import org.apache.storm.trident.state.State; -import org.apache.storm.trident.tuple.TridentTuple; -import org.elasticsearch.client.Request; -import org.elasticsearch.client.Response; -import org.elasticsearch.client.RestClient; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Trident State for storing tuple to ES document. - * @since 0.11 - */ -class EsState implements State { - private static final Logger LOG = LoggerFactory.getLogger(EsState.class); - private static RestClient client; - private EsConfig esConfig; - private final ObjectMapper objectMapper; - private EsTupleMapper tupleMapper; - - /** - * EsState constructor. - * - * @param esConfig Elasticsearch configuration containing node addresses and cluster name {@link EsConfig} - * @param tupleMapper Tuple to ES document mapper {@link EsTupleMapper} - */ - EsState(EsConfig esConfig, EsTupleMapper tupleMapper) { - this.esConfig = esConfig; - this.objectMapper = new ObjectMapper(); - this.tupleMapper = tupleMapper; - } - - /** - * Elasticsearch index requests with same id will result in update operation - * which means if same tuple replays, only one record will be stored in elasticsearch for same document - * without control with txid. - */ - @Override - public void beginCommit(Long txid) { - - } - - /** - * Elasticsearch index requests with same id will result in update operation - * which means if same tuple replays, only one record will be stored in elasticsearch for same document - * without control with txid. - */ - @Override - public void commit(Long txid) { - - } - - public void prepare() { - try { - synchronized (EsState.class) { - if (client == null) { - client = new StormElasticSearchClient(esConfig).construct(); - } - } - } catch (Exception e) { - LOG.warn("unable to initialize EsState ", e); - } - } - - private String buildRequest(List tuples) throws JsonProcessingException { - StringBuilder bulkRequest = new StringBuilder(); - for (TridentTuple tuple : tuples) { - String source = tupleMapper.getSource(tuple); - String index = tupleMapper.getIndex(tuple); - String type = tupleMapper.getType(tuple); - String id = tupleMapper.getId(tuple); - - IndexDoc indexDoc = new IndexDoc(new Index(index, type, id)); - SourceDoc sourceDoc = new SourceDoc(source); - bulkRequest.append(objectMapper.writeValueAsString(indexDoc)).append('\n'); - bulkRequest.append(objectMapper.writeValueAsString(sourceDoc)).append('\n'); - - } - return bulkRequest.toString(); - } - - /** - * Store current state to ElasticSearch. - * - * @param tuples list of tuples for storing to ES. - * Each tuple should have relevant fields (source, index, type, id) for EsState's tupleMapper to extract ES document. - */ - public void updateState(List tuples) { - try { - String bulkRequest = buildRequest(tuples); - - final Request request = new Request("post", "_bulk"); - request.setEntity(new StringEntity(bulkRequest)); - Response response = client.performRequest(request); - BulkIndexResponse bulkResponse = objectMapper.readValue(response.getEntity().getContent(), BulkIndexResponse.class); - if (bulkResponse.hasErrors()) { - LOG.warn("failed processing bulk index requests: " + bulkResponse.getFirstError() + ": " + bulkResponse.getFirstResult()); - throw new FailedException(); - } - } catch (IOException e) { - LOG.warn("failed processing bulk index requests: " + e.toString()); - throw new FailedException(e); - } - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsStateFactory.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsStateFactory.java deleted file mode 100644 index d79ec2e70bf..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsStateFactory.java +++ /dev/null @@ -1,56 +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.storm.elasticsearch.trident; - -import static java.util.Objects.requireNonNull; - -import java.util.Map; - -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTupleMapper; -import org.apache.storm.task.IMetricsContext; -import org.apache.storm.trident.state.State; -import org.apache.storm.trident.state.StateFactory; - -/** - * StateFactory for providing EsState. - * @since 0.11 - */ -public class EsStateFactory implements StateFactory { - private final EsConfig esConfig; - private final EsTupleMapper tupleMapper; - - /** - * EsStateFactory constructor. - * - * @param esConfig Elasticsearch configuration containing node addresses and cluster name {@link EsConfig} - * @param tupleMapper Tuple to ES document mapper {@link EsTupleMapper} - */ - public EsStateFactory(EsConfig esConfig, EsTupleMapper tupleMapper) { - this.esConfig = requireNonNull(esConfig); - this.tupleMapper = requireNonNull(tupleMapper); - } - - @Override - public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) { - EsState esState = new EsState(esConfig, tupleMapper); - esState.prepare(); - return esState; - } -} diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsUpdater.java b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsUpdater.java deleted file mode 100644 index 30f296c1937..00000000000 --- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsUpdater.java +++ /dev/null @@ -1,36 +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.storm.elasticsearch.trident; - -import java.util.List; - -import org.apache.storm.trident.operation.TridentCollector; -import org.apache.storm.trident.state.BaseStateUpdater; -import org.apache.storm.trident.tuple.TridentTuple; - -public class EsUpdater extends BaseStateUpdater { - /** - * {@inheritDoc} - * Each tuple should have relevant fields (source, index, type, id) for EsState's tupleMapper to extract ES document. - */ - @Override - public void updateState(EsState state, List tuples, TridentCollector collector) { - state.updateState(tuples); - } -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltIntegrationTest.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltIntegrationTest.java deleted file mode 100644 index 0819f8a8e86..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltIntegrationTest.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.storm.elasticsearch.bolt; - -import org.apache.storm.elasticsearch.common.EsTestUtil; -import org.apache.storm.testing.IntegrationTest; -import org.elasticsearch.action.admin.indices.create.CreateIndexRequest; -import org.elasticsearch.client.RequestOptions; -import org.elasticsearch.node.Node; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.BeforeEach; -import org.testcontainers.elasticsearch.ElasticsearchContainer; - -import java.io.IOException; - -@IntegrationTest -public abstract class AbstractEsBoltIntegrationTest extends AbstractEsBoltTest { - - protected static ElasticsearchContainer node; - - @BeforeAll - public static void startElasticSearchNode() throws Exception { - node = EsTestUtil.startEsNode(); - EsTestUtil.ensureEsGreen(node); - } - - @AfterAll - public static void closeElasticSearchNode() throws Exception { - EsTestUtil.stopEsNode(node); - } - - @BeforeEach - public void createIndex() throws IOException { - EsTestUtil.getRestHighLevelClient(node).indices().create(new CreateIndexRequest(index), RequestOptions.DEFAULT); - } - - @AfterEach - public void clearIndex() throws Exception { - EsTestUtil.clearIndex(node, index); - EsTestUtil.clearIndex(node, "missing"); - } -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltTest.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltTest.java deleted file mode 100644 index 594483121bd..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltTest.java +++ /dev/null @@ -1,78 +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.storm.elasticsearch.bolt; - -import com.google.common.testing.NullPointerTester; - -import java.lang.reflect.Method; -import java.util.UUID; - -import org.apache.storm.Config; -import org.apache.storm.task.OutputCollector; -import org.apache.storm.elasticsearch.common.EsConfig; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.Mock; -import org.mockito.junit.jupiter.MockitoExtension; - -@ExtendWith(MockitoExtension.class) -public abstract class AbstractEsBoltTest { - - protected static Config config = new Config(); - protected static final String documentId = UUID.randomUUID().toString(); - protected static final String index = "index"; - protected static final String source = "{\"user\":\"user1\"}"; - protected static final String type = "type"; - - @Mock - protected OutputCollector outputCollector; - - protected Bolt bolt; - - @BeforeEach - public void createBolt() throws Exception { - bolt = createBolt(esConfig()); - bolt.prepare(config, null, outputCollector); - } - - protected abstract Bolt createBolt(EsConfig esConfig); - - protected EsConfig esConfig() { - return new EsConfig(); - } - - @AfterEach - public void cleanupBolt() throws Exception { - bolt.cleanup(); - } - - @Test - public void constructorsThrowOnNull() throws Exception { - new NullPointerTester().setDefault(EsConfig.class, esConfig()).testAllPublicConstructors(getBoltClass()); - } - - @Test - public void getEndpointThrowsOnNull() throws Exception { - Method getEndpointMethod = AbstractEsBolt.class.getDeclaredMethod("getEndpoint", String.class, String.class, String.class); - new NullPointerTester().setDefault(String.class, "test").testMethodParameter(null, getEndpointMethod, 0); - } - - protected abstract Class getBoltClass(); -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexBoltTest.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexBoltTest.java deleted file mode 100644 index a7ccee4bf27..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexBoltTest.java +++ /dev/null @@ -1,98 +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.storm.elasticsearch.bolt; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.mockito.Mockito.verify; - -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTestUtil; -import org.apache.storm.tuple.Tuple; -import org.elasticsearch.action.admin.indices.refresh.RefreshRequest; -import org.elasticsearch.action.search.SearchRequest; -import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.client.RequestOptions; -import org.elasticsearch.client.RestHighLevelClient; -import org.elasticsearch.index.query.QueryBuilders; -import org.elasticsearch.index.query.TermQueryBuilder; -import org.elasticsearch.search.builder.SearchSourceBuilder; -import org.junit.jupiter.api.Test; - -import java.io.IOException; - -public class EsIndexBoltTest extends AbstractEsBoltIntegrationTest { - - @Test - public void testEsIndexBolt() throws IOException { - Tuple tuple = createTestTuple(index, type); - - bolt.execute(tuple); - - verify(outputCollector).ack(tuple); - - RestHighLevelClient client = EsTestUtil.getRestHighLevelClient(node); - RefreshRequest request = new RefreshRequest(index); - client.indices().refresh(request, RequestOptions.DEFAULT); - - SearchRequest searchRequest = new SearchRequest(index); - SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); - searchSourceBuilder.query(new TermQueryBuilder("_type", type)); - searchSourceBuilder.size(0); - searchRequest.source(searchSourceBuilder); - - SearchResponse resp = client.search(searchRequest, RequestOptions.DEFAULT); - - assertEquals(1, resp.getHits().getTotalHits()); - } - - @Test - public void indexMissing() throws IOException { - String index = "missing"; - - Tuple tuple = createTestTuple(index, type); - - bolt.execute(tuple); - - verify(outputCollector).ack(tuple); - - RestHighLevelClient client = EsTestUtil.getRestHighLevelClient(node); - SearchRequest searchRequest = new SearchRequest(index); - SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); - searchSourceBuilder.query(new TermQueryBuilder("_type", type)); - searchSourceBuilder.size(0); - searchRequest.source(searchSourceBuilder); - - SearchResponse resp = client.search(searchRequest, RequestOptions.DEFAULT); - - assertEquals(1, resp.getHits().getTotalHits()); - } - - private Tuple createTestTuple(String index, String type) { - return EsTestUtil.generateTestTuple(source, index, type, documentId); - } - - @Override - protected EsIndexBolt createBolt(EsConfig esConfig) { - return new EsIndexBolt(esConfig); - } - - @Override - protected Class getBoltClass() { - return EsIndexBolt.class; - } -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltIntegrationTest.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltIntegrationTest.java deleted file mode 100644 index c31e6f31092..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltIntegrationTest.java +++ /dev/null @@ -1,112 +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.storm.elasticsearch.bolt; - -import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; - -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTestUtil; -import org.apache.storm.tuple.Tuple; -import org.apache.storm.tuple.Values; -import org.elasticsearch.action.index.IndexRequest; -import org.elasticsearch.client.RequestOptions; -import org.elasticsearch.client.ResponseException; -import org.elasticsearch.client.RestHighLevelClient; -import org.elasticsearch.xcontent.XContentType; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.ArgumentCaptor; -import org.mockito.Captor; -import org.mockito.junit.jupiter.MockitoExtension; - -import java.io.IOException; - -@ExtendWith(MockitoExtension.class) -public class EsLookupBoltIntegrationTest extends AbstractEsBoltIntegrationTest { - - @Captor - private ArgumentCaptor anchor; - - @Captor - private ArgumentCaptor emittedValues; - - private final Tuple tuple = EsTestUtil.generateTestTuple(source, index, type, documentId); - - @Override - protected EsLookupBolt createBolt(EsConfig esConfig) { - return new EsLookupBolt(esConfig); - } - - @BeforeEach - public void populateIndexWithTestData() throws IOException { - IndexRequest indexRequest = new IndexRequest(index, type, documentId) - .source(source, XContentType.JSON); - RestHighLevelClient client = EsTestUtil.getRestHighLevelClient(node); - client.index(indexRequest, RequestOptions.DEFAULT); - } - - @Test - public void anchorsTheTuple() { - bolt.execute(tuple); - - verify(outputCollector).emit(anchor.capture(), emittedValues.capture()); - assertThat(anchor.getValue(), is(tuple)); - } - - @Test - public void emitsExpectedValues() { - Values expectedValues = expectedValues(); - - bolt.execute(tuple); - - verify(outputCollector).emit(anchor.capture(), emittedValues.capture()); - assertThat(emittedValues.getValue(), is(expectedValues)); - } - - @Test - public void acksTuple() { - bolt.execute(tuple); - - verify(outputCollector).ack(anchor.capture()); - assertThat(anchor.getValue(), is(tuple)); - } - - @Test - public void indexMissing() { - Tuple tuple = EsTestUtil.generateTestTuple(source, "missing", type, documentId); - bolt.execute(tuple); - - verify(outputCollector, never()).emit(any(Tuple.class), any(Values.class)); - verify(outputCollector).reportError(any(ResponseException.class)); - verify(outputCollector).fail(tuple); - } - - private Values expectedValues() { - return new Values(index, type, documentId, source); - } - - @Override - protected Class getBoltClass() { - return EsLookupBolt.class; - } -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltTest.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltTest.java deleted file mode 100644 index 0067e465d48..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltTest.java +++ /dev/null @@ -1,129 +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.storm.elasticsearch.bolt; - -import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -import java.io.IOException; -import java.util.HashMap; -import java.util.Map; -import java.util.UUID; - -import org.apache.storm.elasticsearch.EsLookupResultOutput; -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTestUtil; -import org.apache.storm.elasticsearch.common.EsTupleMapper; -import org.apache.storm.topology.OutputFieldsDeclarer; -import org.apache.storm.tuple.Fields; -import org.apache.storm.tuple.Tuple; -import org.elasticsearch.client.Request; -import org.elasticsearch.client.Response; -import org.elasticsearch.client.ResponseException; -import org.elasticsearch.client.RestClient; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.ArgumentCaptor; -import org.mockito.Mock; -import org.mockito.junit.jupiter.MockitoExtension; -import org.mockito.junit.jupiter.MockitoSettings; -import org.mockito.quality.Strictness; - -@ExtendWith(MockitoExtension.class) -@MockitoSettings(strictness = Strictness.WARN) -public class EsLookupBoltTest extends AbstractEsBoltTest { - - static final Map params = new HashMap<>(); - - @Mock - private EsTupleMapper tupleMapper; - - @Mock - private EsLookupResultOutput output; - - @Mock - private Tuple tuple; - - @Mock - private RestClient client; - - private RestClient originalClient; - - @Override - protected EsLookupBolt createBolt(EsConfig esConfig) { - originalClient = EsLookupBolt.getClient(); - EsLookupBolt.replaceClient(this.client); - return new EsLookupBolt(esConfig, tupleMapper, output); - } - - @AfterEach - public void replaceClientWithOriginal() { - EsLookupBolt.replaceClient(originalClient); - } - - @BeforeEach - public void configureBoltDependencies() { - when(tupleMapper.getIndex(tuple)).thenReturn(index); - when(tupleMapper.getType(tuple)).thenReturn(type); - when(tupleMapper.getId(tuple)).thenReturn(documentId); - } - - private void makeRequestAndThrow(Exception exception) throws IOException { - final Request request = new Request("get", AbstractEsBolt.getEndpoint(index, type, documentId)); - request.addParameters(params); - when(client.performRequest(request)).thenThrow(exception); - bolt.execute(tuple); - } - - @Test - public void failsTupleWhenClientThrows() throws Exception { - makeRequestAndThrow(EsTestUtil.generateMockResponseException()); - verify(outputCollector).fail(tuple); - } - - @Test - public void reportsExceptionWhenClientThrows() throws Exception { - ResponseException responseException = EsTestUtil.generateMockResponseException(); - makeRequestAndThrow(responseException); - - verify(outputCollector).reportError(responseException); - } - - @Test - public void fieldsAreDeclaredThroughProvidedOutput() { - Fields fields = new Fields(UUID.randomUUID().toString()); - when(output.fields()).thenReturn(fields); - OutputFieldsDeclarer declarer = mock(OutputFieldsDeclarer.class); - bolt.declareOutputFields(declarer); - - ArgumentCaptor declaredFields = ArgumentCaptor.forClass(Fields.class); - verify(declarer).declare(declaredFields.capture()); - - assertThat(declaredFields.getValue(), is(fields)); - } - - @Override - protected Class getBoltClass() { - return EsLookupBolt.class; - } -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsPercolateBoltTest.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsPercolateBoltTest.java deleted file mode 100644 index d50127db08e..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsPercolateBoltTest.java +++ /dev/null @@ -1,104 +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.storm.elasticsearch.bolt; - -import static org.hamcrest.CoreMatchers.instanceOf; -import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; - -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTestUtil; -import org.apache.storm.elasticsearch.response.PercolateResponse; -import org.apache.storm.tuple.Tuple; -import org.apache.storm.tuple.Values; -import org.elasticsearch.action.index.IndexRequest; -import org.elasticsearch.action.support.WriteRequest; -import org.elasticsearch.client.RequestOptions; -import org.elasticsearch.client.ResponseException; -import org.elasticsearch.client.RestHighLevelClient; -import org.elasticsearch.xcontent.XContentType; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.mockito.ArgumentCaptor; - -import java.io.IOException; - -public class EsPercolateBoltTest extends AbstractEsBoltIntegrationTest { - - private final String source = "{\"user\":\"user1\"}"; - - @Override - protected EsPercolateBolt createBolt(EsConfig esConfig) { - return new EsPercolateBolt(esConfig); - } - - @BeforeEach - public void populateIndexWithTestData() throws IOException { - IndexRequest indexRequest = new IndexRequest(index, ".percolator", documentId) - .source(source, XContentType.JSON) - .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); // setRefresh(true) in Elasticsearch 6.x - - RestHighLevelClient client = EsTestUtil.getRestHighLevelClient(node); - - client.index(indexRequest, RequestOptions.DEFAULT); - } - - @Test - public void testEsPercolateBolt() { - Tuple tuple = EsTestUtil.generateTestTuple(source, index, type, null); - - bolt.execute(tuple); - - verify(outputCollector).ack(tuple); - ArgumentCaptor emitCaptor = ArgumentCaptor.forClass(Values.class); - verify(outputCollector).emit(emitCaptor.capture()); - assertThat(emitCaptor.getValue().get(0), is(source)); - assertThat(emitCaptor.getValue().get(1), instanceOf(PercolateResponse.Match.class)); - } - - @Test - public void noDocumentsMatch() { - Tuple tuple = EsTestUtil.generateTestTuple("{\"user\":\"user2\"}", index, type, null); - - bolt.execute(tuple); - - verify(outputCollector).ack(tuple); - verify(outputCollector, never()).emit(any(Values.class)); - } - - @Test - public void indexMissing() { - String index = "missing"; - - Tuple tuple = EsTestUtil.generateTestTuple(source, index, type, null); - - bolt.execute(tuple); - - verify(outputCollector, never()).emit(any(Values.class)); - verify(outputCollector).reportError(any(ResponseException.class)); - verify(outputCollector).fail(tuple); - } - - @Override - protected Class getBoltClass() { - return EsPercolateBolt.class; - } -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsConfigTest.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsConfigTest.java deleted file mode 100644 index b2e2df91ffe..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsConfigTest.java +++ /dev/null @@ -1,68 +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.storm.elasticsearch.common; - -import static org.junit.jupiter.api.Assertions.assertEquals; - -import org.apache.http.HttpHost; - -import com.google.common.testing.NullPointerTester; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; - -public class EsConfigTest { - - @Test - public void urlsCannotBeEmpty() { - Assertions.assertThrows(IllegalArgumentException.class, () -> new EsConfig(new String[] {})); - } - - @Test - public void constructorThrowsOnNull() { - new NullPointerTester().testAllPublicConstructors(EsConfig.class); - } - - @Test - public void usesElasticsearchDefaults() { - EsConfig esConfig = new EsConfig(); - HttpHost[] httpHosts = esConfig.getHttpHosts(); - assertEquals(1, httpHosts.length); - assertEquals("http", httpHosts[0].getSchemeName()); - assertEquals(9200, httpHosts[0].getPort()); - assertEquals("localhost", httpHosts[0].getHostName()); - } - - @Test - public void setsSchemePortAndHost() { - EsConfig esConfig = new EsConfig("https://somehost:1234"); - HttpHost[] httpHosts = esConfig.getHttpHosts(); - assertEquals(1, httpHosts.length); - assertEquals("https", httpHosts[0].getSchemeName()); - assertEquals(1234, httpHosts[0].getPort()); - assertEquals("somehost", httpHosts[0].getHostName()); - } - - @Test - public void usesMultipleHostnames() { - EsConfig esConfig = new EsConfig("http://host1:9200", "http://host2:9200"); - HttpHost[] httpHosts = esConfig.getHttpHosts(); - assertEquals(2, httpHosts.length); - assertEquals("host1", httpHosts[0].getHostName()); - assertEquals("host2", httpHosts[1].getHostName()); - } -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsConstants.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsConstants.java deleted file mode 100644 index 98bb71dd46d..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsConstants.java +++ /dev/null @@ -1,22 +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.storm.elasticsearch.common; - -public class EsConstants { - public static String clusterName = "test-cluster"; -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsTestUtil.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsTestUtil.java deleted file mode 100644 index 303fda63578..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsTestUtil.java +++ /dev/null @@ -1,128 +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.storm.elasticsearch.common; - -import static org.hamcrest.CoreMatchers.equalTo; -import static org.hamcrest.MatcherAssert.assertThat; - -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import java.io.File; -import java.io.IOException; -import java.util.HashMap; - -import org.apache.commons.io.FileUtils; -import org.apache.http.RequestLine; -import org.apache.http.StatusLine; -import org.apache.storm.Config; -import org.apache.storm.task.GeneralTopologyContext; -import org.apache.storm.topology.TopologyBuilder; -import org.apache.storm.trident.tuple.TridentTuple; -import org.apache.storm.trident.tuple.TridentTupleView; -import org.apache.storm.tuple.Fields; -import org.apache.storm.tuple.Tuple; -import org.apache.storm.tuple.TupleImpl; -import org.apache.storm.tuple.Values; -import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest; -import org.elasticsearch.client.*; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.testcontainers.elasticsearch.ElasticsearchContainer; - -public class EsTestUtil { - private static final Logger LOG = LoggerFactory.getLogger(EsTestUtil.class); - - public static Tuple generateTestTuple(String source, String index, String type, String id) { - TopologyBuilder builder = new TopologyBuilder(); - GeneralTopologyContext topologyContext = new GeneralTopologyContext(builder.createTopology(), - new Config(), new HashMap<>(), new HashMap<>(), new HashMap<>(), "") { - @Override - public Fields getComponentOutputFields(String componentId, String streamId) { - return new Fields("source", "index", "type", "id"); - } - }; - return new TupleImpl(topologyContext, new Values(source, index, type, id), source, 1, ""); - } - - public static TridentTuple generateTestTridentTuple(String source, String index, String type, String id) { - Fields fields = new Fields("source", "index", "type", "id"); - return TridentTupleView.createFreshTuple(fields, new Values(source, index, type, id)); - } - - /** - * Generate a mock {@link ResponseException}. - * @return a mock {@link ResponseException}. - * @throws IOException - */ - public static ResponseException generateMockResponseException() throws IOException { - Response response = mock(Response.class); - RequestLine requestLine = mock(RequestLine.class); - StatusLine statusLine = mock(StatusLine.class); - when(response.getRequestLine()).thenReturn(requestLine); - when(response.getStatusLine()).thenReturn(statusLine); - return new ResponseException(response); - } - - public static RestHighLevelClient getRestHighLevelClient(ElasticsearchContainer node) { - final EsConfig cfg = new EsConfig(node.getHttpHostAddress()); - return new RestHighLevelClientBuilder(new StormElasticSearchClient(cfg).construct()).build(); - } - - public static ElasticsearchContainer startEsNode() { - String version = System.getProperty("elasticsearch-version"); - if (version == null) version = "7.17.13"; - LOG.info("Starting docker instance of Elasticsearch {}...", version); - - final ElasticsearchContainer container = - new ElasticsearchContainer( - "docker.elastic.co/elasticsearch/elasticsearch:" + version); - container.start(); - return container; - } - - public static void ensureEsGreen(ElasticsearchContainer node) { - assertThat("cluster status is green", node.isHealthy()); - } - - /** - * Stop the given Elasticsearch node and clear the data directory. - * @param node - */ - public static void stopEsNode(ElasticsearchContainer node) { - node.close(); - try { - FileUtils.deleteDirectory(new File("./data")); - } catch (IOException e) { - LOG.warn(e.toString()); - } - } - - /** - * Clear the given index if it exists - * @param node - the node to connect to - * @param index - the index to clear - */ - public static void clearIndex(ElasticsearchContainer node, String index) { - try { - getRestHighLevelClient(node).indices().delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT); - } catch (IOException ignore) { - - } - } -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/EsStateFactoryTest.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/EsStateFactoryTest.java deleted file mode 100644 index a1b4fe80aa3..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/EsStateFactoryTest.java +++ /dev/null @@ -1,32 +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.storm.elasticsearch.trident; - -import com.google.common.testing.NullPointerTester; - -import org.apache.storm.elasticsearch.common.EsConfig; -import org.junit.jupiter.api.Test; - -public class EsStateFactoryTest { - - @Test - public void constructorThrowsOnNull() throws Exception { - new NullPointerTester().setDefault(EsConfig.class, new EsConfig(new String[] {"http://localhost:9300"})) - .testAllPublicConstructors(EsStateFactory.class); - } -} diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/EsStateTest.java b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/EsStateTest.java deleted file mode 100644 index 8e86e7dad8a..00000000000 --- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/EsStateTest.java +++ /dev/null @@ -1,97 +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.storm.elasticsearch.trident; - -import java.util.ArrayList; -import java.util.List; -import java.util.UUID; - -import org.apache.storm.elasticsearch.common.DefaultEsTupleMapper; -import org.apache.storm.elasticsearch.common.EsConfig; -import org.apache.storm.elasticsearch.common.EsTestUtil; -import org.apache.storm.testing.IntegrationTest; -import org.apache.storm.trident.tuple.TridentTuple; -import org.elasticsearch.node.Node; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.junit.jupiter.MockitoExtension; -import org.testcontainers.elasticsearch.ElasticsearchContainer; - -@IntegrationTest -@ExtendWith(MockitoExtension.class) -public class EsStateTest { - - private static ElasticsearchContainer node; - - private final String[] documentId = {UUID.randomUUID().toString(), UUID.randomUUID().toString()}; - private final String index = "index"; - private final String type = "type"; - private final String[] source = {"{\"user\":\"user1\"}", "{\"user\":\"user1\"}"}; - - private EsState state = createEsState(); - - @BeforeAll - public static void startElasticSearchNode() throws Exception { - node = EsTestUtil.startEsNode(); - EsTestUtil.ensureEsGreen(node); - } - - @AfterAll - public static void closeElasticSearchNode() throws Exception { - EsTestUtil.stopEsNode(node); - } - - @AfterEach - public void clearIndex() throws Exception { - EsTestUtil.clearIndex(node, index); - EsTestUtil.clearIndex(node, "missing"); - } - - private EsState createEsState() { - EsState state = new EsState(esConfig(), new DefaultEsTupleMapper()); - state.prepare(); - return state; - } - - private EsConfig esConfig() { - return new EsConfig(); - } - - private List tuples(String index, String type, String[] ids, String[] sources) { - List tuples = new ArrayList<>(); - for (int i = 0; i < ids.length; i++) { - tuples.add(EsTestUtil.generateTestTridentTuple(sources[i], index, type, ids[i])); - } - return tuples; - } - - @Test - public void updateState() throws Exception { - List tuples = tuples(index, type, documentId, source); - state.updateState(tuples); - } - - @Test - public void indexMissing() throws Exception { - List tuples = tuples("missing", type, documentId, source); - state.updateState(tuples); - } -} diff --git a/external/storm-elasticsearch/src/test/resources/log4j2.xml b/external/storm-elasticsearch/src/test/resources/log4j2.xml deleted file mode 100644 index b03ad813dac..00000000000 --- a/external/storm-elasticsearch/src/test/resources/log4j2.xml +++ /dev/null @@ -1,33 +0,0 @@ - - - - - - - %-4r [%t] %-5level %c{1} - %msg%n - - - - - - - - - - - \ No newline at end of file diff --git a/pom.xml b/pom.xml index 046b531e333..f144d47651d 100644 --- a/pom.xml +++ b/pom.xml @@ -127,7 +127,6 @@ 4.2.2 2.2.2 2.2 - 7.17.13 1.16.0 5.1.0 5.18.3 @@ -495,7 +494,6 @@ external/storm-hive external/storm-jdbc external/storm-redis - external/storm-elasticsearch external/storm-metrics external/storm-metrics-prometheus external/storm-kafka-client @@ -523,7 +521,6 @@ examples/storm-jdbc-examples examples/storm-hdfs-examples examples/storm-hive-examples - examples/storm-elasticsearch-examples examples/storm-jms-examples examples/storm-perf diff --git a/storm-dist/binary/final-package/src/main/assembly/binary.xml b/storm-dist/binary/final-package/src/main/assembly/binary.xml index 03e448d3f96..602dde8057c 100644 --- a/storm-dist/binary/final-package/src/main/assembly/binary.xml +++ b/storm-dist/binary/final-package/src/main/assembly/binary.xml @@ -145,13 +145,6 @@ README.* - - ${project.basedir}/../../../external/storm-elasticsearch - external/storm-elasticsearch - - README.* - - ${project.basedir}/../../../external/flux