Appying for Elasticsearch 2.3.5

external/storm-elasticsearch/pom.xml

@@ -21,7 +21,7 @@
     <parent>
         <artifactId>storm</artifactId>
         <groupId>org.apache.storm</groupId>
-        <version>1.0.3-SNAPSHOT</version>
+        <version>1.0.2</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
 
@@ -36,7 +36,7 @@
     </developers>
 
     <properties>
-        <elasticsearch.version>1.6.0</elasticsearch.version>
+        <elasticsearch.version>2.3.5</elasticsearch.version>
     </properties>
 
     <dependencies>

external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/AbstractEsBolt.java

@@ -21,18 +21,17 @@
 
 import org.apache.storm.elasticsearch.common.EsConfig;
 import org.apache.storm.elasticsearch.common.StormElasticSearchClient;
-import org.elasticsearch.client.Client;
-import org.elasticsearch.common.annotations.VisibleForTesting;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
 import org.apache.storm.task.OutputCollector;
 import org.apache.storm.task.TopologyContext;
 import org.apache.storm.topology.OutputFieldsDeclarer;
 import org.apache.storm.topology.base.BaseRichBolt;
 import org.apache.storm.tuple.Tuple;
+import org.elasticsearch.client.Client;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import static org.elasticsearch.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkNotNull;
 
 public abstract class AbstractEsBolt extends BaseRichBolt {
 

external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsIndexBolt.java

@@ -26,7 +26,7 @@
 
 import java.util.Map;
 
-import static org.elasticsearch.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
  * Basic bolt for storing tuple to ES document.

external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupBolt.java

@@ -29,7 +29,7 @@
 import org.apache.storm.tuple.Tuple;
 import org.apache.storm.tuple.Values;
 
-import static org.elasticsearch.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
  * @since 0.11

external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsPercolateBolt.java

@@ -30,7 +30,7 @@
 
 import java.util.Map;
 
-import static org.elasticsearch.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
  * Basic bolt for retrieve matched percolate queries.

external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/EsConfig.java

@@ -22,11 +22,10 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.settings.Settings;
 
-import static org.elasticsearch.common.base.Preconditions.checkArgument;
-import static org.elasticsearch.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
  * @since 0.11
@@ -74,7 +73,7 @@ TransportAddresses getTransportAddresses() {
     }
 
     Settings toBasicSettings() {
-        return ImmutableSettings.settingsBuilder()
+        return Settings.settingsBuilder()
                                 .put("cluster.name", clusterName)
                                 .put(additionalConfiguration)
                                 .build();

external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/StormElasticSearchClient.java

@@ -34,7 +34,7 @@ public StormElasticSearchClient(EsConfig esConfig) {
 
     public Client construct() {
         Settings settings = esConfig.toBasicSettings();
-        TransportClient transportClient = new TransportClient(settings);
+        TransportClient transportClient = TransportClient.builder().settings(settings).build();
         addTransportAddresses(transportClient);
         return transportClient;
     }

external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/common/TransportAddresses.java

@@ -17,6 +17,7 @@
  */
 package org.apache.storm.elasticsearch.common;
 
+import java.net.InetSocketAddress;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
@@ -59,7 +60,7 @@ private InetSocketTransportAddress transformToInetAddress(String node) {
                     "Incorrect Elasticsearch node format, should follow {host}" + DELIMETER + "{port} pattern");
         }
         String hostname = hostname(hostAndPort[0]);
-        return new InetSocketTransportAddress(hostname, port(hostAndPort[1]));
+        return new InetSocketTransportAddress(new InetSocketAddress(hostname, port(hostAndPort[1])));
     }
 
     private String hostname(String input) {

external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsStateFactory.java

@@ -26,7 +26,7 @@
 
 import java.util.Map;
 
-import static org.elasticsearch.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
  * StateFactory for providing EsState.

external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltIntegrationTest.java

@@ -20,12 +20,12 @@
 import org.apache.storm.testing.IntegrationTest;
 import org.apache.commons.io.FileUtils;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
-import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.client.Requests;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Priority;
-import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.node.Node;
@@ -60,20 +60,20 @@ public static void startElasticSearchNode() throws Exception {
         Thread.sleep(1000);
     }
 
-    private static ImmutableSettings.Builder createSettings() {
-        return ImmutableSettings.builder()
+    private static Settings.Builder createSettings() {
+        return Settings.builder()
                                 .put(ClusterName.SETTING, "test-cluster")
                                 .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
                                 .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
                                 .put(EsExecutors.PROCESSORS, 1)
                                 .put("http.enabled", false)
                                 .put("index.percolator.map_unmapped_fields_as_string", true)
-                                .put("index.store.type", "memory");
+                                .put("index.store.type", "mmapfs")
+                                .put("path.home", "/");
     }
 
     @AfterClass
     public static void closeElasticSearchNode() throws Exception {
-        node.stop();
         node.close();
         FileUtils.deleteDirectory(new File("./data"));
     }

external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/EsTestUtil.java

@@ -27,7 +27,7 @@
 import org.apache.storm.tuple.Values;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
-import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.node.NodeBuilder;
@@ -53,7 +53,7 @@ public static EsTupleMapper generateDefaultTupleMapper() {
 
     public static Node startEsNode(){
         Node node = NodeBuilder.nodeBuilder().data(true).settings(
-                ImmutableSettings.builder()
+                Settings.builder()
                         .put(ClusterName.SETTING, EsConstants.clusterName)
                         .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
                         .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)

external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/common/TransportAddressesTest.java

@@ -23,29 +23,31 @@
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.junit.Assert.assertThat;
 
+import java.net.InetSocketAddress;
+
 public class TransportAddressesTest {
 
     @Test
     public void readsMultipleHosts() throws Exception {
-        String[] hosts = new String[] {"h1:1000", "h2:10003"};
+        String[] hosts = new String[] {"localhost:1000", "localhost:10003"};
         TransportAddresses addresses = new TransportAddresses(hosts);
-        assertThat(addresses, containsInAnyOrder(new InetSocketTransportAddress("h1", 1000),
-                                                 new InetSocketTransportAddress("h2", 10003)));
+        assertThat(addresses, containsInAnyOrder(new InetSocketTransportAddress(new InetSocketAddress("localhost", 1000)),
+                                                 new InetSocketTransportAddress(new InetSocketAddress("localhost", 10003))));
     }
 
     @Test
     public void stripsSpaces() throws Exception {
-        String[] hosts = new String[] {"h1:1000", " h2:10003 "};
+        String[] hosts = new String[] {"localhost:1000", "localhost:10003 "};
         TransportAddresses addresses = new TransportAddresses(hosts);
-        assertThat(addresses, containsInAnyOrder(new InetSocketTransportAddress("h1", 1000),
-                                                 new InetSocketTransportAddress("h2", 10003)));
+        assertThat(addresses, containsInAnyOrder(new InetSocketTransportAddress(new InetSocketAddress("localhost", 1000)),
+                                                 new InetSocketTransportAddress(new InetSocketAddress("localhost", 10003))));
     }
 
     @Test
     public void readsOneHost() throws Exception {
-        String[] hosts = new String[] {"h1:1000"};
+        String[] hosts = new String[] {"localhost:1000"};
         TransportAddresses addresses = new TransportAddresses(hosts);
-        assertThat(addresses, containsInAnyOrder(new InetSocketTransportAddress("h1", 1000)));
+        assertThat(addresses, containsInAnyOrder(new InetSocketTransportAddress(new InetSocketAddress("localhost", 1000))));
     }
 
     @Test(expected = IllegalArgumentException.class)

pom.xml

@@ -223,7 +223,7 @@
         <clojure.tools.cli.version>0.2.4</clojure.tools.cli.version>
         <disruptor.version>3.3.2</disruptor.version>
         <jgrapht.version>0.9.0</jgrapht.version>
-        <guava.version>16.0.1</guava.version>
+        <guava.version>19.0</guava.version>
         <netty.version>3.9.0.Final</netty.version>
         <log4j-over-slf4j.version>1.6.6</log4j-over-slf4j.version>
         <log4j.version>2.1</log4j.version>