Merge branch 'disable_cluster_nodes_sniffing' of https://github.com/centic9/elasticsearch-reindex-tool into centic9-disable_cluster_nodes_sniffing

Author: awislowski

README.md

@@ -86,6 +86,10 @@ Options:
        Target f.e. http://localhost:9300/target_index/type
     -tc, target-cluster
        Target cluster name
+    -disable-cluster-sniffing
+       Don't try to determine additional cluster nodes (e.g. when your network
+       only allows access to one of the nodes)
+       Default: false
     -segmentationField
        Segmentation field
     -segmentationPrefixes
@@ -102,6 +106,10 @@ Options:
 `segmentationField`, `segmentationThreshold` and `segmentationPrefixes` are optional parameters, allowing to spread
 querying for field with double values or prefix for string field
 
+`disable-cluster-sniffing` allows to work in cases where the network-setup makes it impossible to connect to all nodes
+of the source or target cluster. Note that it may lead to slightly reduced reindexing rates as data can only be sent
+via one node then.
+
 During reindex process progress message is prompted after each scroll query.
 
 Example of progress message with the time how long it lasts, number of items queried and indexed, occupancy of queue, number of concurrent reader threads and number of failed document indexing:

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/ReindexCommandParser.java

@@ -2,11 +2,10 @@
 
 import com.beust.jcommander.JCommander;
 import com.beust.jcommander.ParameterException;
+
 import pl.allegro.tech.search.elasticsearch.tools.reindex.command.ReindexCommand;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointer;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointerBuilder;
-import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchQuery;
-import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchQueryBuilder;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ParsingElasticsearchAddressException;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.query.QuerySegmentation;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.query.QuerySegmentationFactory;
@@ -40,10 +39,12 @@ private void buildReindexParameters(ReindexCommand command) {
     sourcePointer = ElasticDataPointerBuilder.builder()
         .setClusterName(command.getSourceClusterName())
         .setAddress(command.getSource())
+        .setSniff(!command.isDisableSniff())
         .build();
     targetPointer = ElasticDataPointerBuilder.builder()
         .setClusterName(command.getTargetClusterName())
         .setAddress(command.getTarget())
+        .setSniff(!command.isDisableSniff())
         .build();
     segmentation = getFieldSegmentation(command);
   }

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/command/ReindexCommand.java

@@ -1,9 +1,9 @@
 package pl.allegro.tech.search.elasticsearch.tools.reindex.command;
 
-import com.beust.jcommander.Parameter;
-
 import java.util.List;
 
+import com.beust.jcommander.Parameter;
+
 public class ReindexCommand {
 
   @Parameter(names = { "-s", "source" }, description = "Source f.e. http://localhost:9300/source_index/type",
@@ -37,6 +37,9 @@ public class ReindexCommand {
   @Parameter(names = { "-segmentationPrefixes" }, description = "Segmentation prefixes (comma-separated)")
   private List<String> segmentationPrefixes;
 
+  @Parameter(names = { "-disable-cluster-sniffing" }, description = "Don't try to determine additional cluster nodes (e.g. when your network only allows access to one of the nodes)")
+  private boolean disableSniff;
+
   public String getSourceClusterName() {
     return sourceClusterName;
   }
@@ -74,4 +77,8 @@ public String getSource() {
   public String getTarget() {
     return target;
   }
+
+  public boolean isDisableSniff() {
+    return disableSniff;
+  }
 }

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/connection/ElasticDataPointer.java

@@ -7,13 +7,15 @@ public class ElasticDataPointer {
   private final String indexName;
   private final String typeName;
   private final int port;
+  private final boolean sniff;
 
-  ElasticDataPointer(String host, String clusterName, String indexName, String typeName, int port) {
+  ElasticDataPointer(String host, String clusterName, String indexName, String typeName, int port, boolean sniff) {
     this.host = host;
     this.clusterName = clusterName;
     this.indexName = indexName;
     this.typeName = typeName;
     this.port = port;
+    this.sniff = sniff;
   }
 
   public String getHost() {
@@ -35,4 +37,8 @@ public String getTypeName() {
   public int getPort() {
     return port;
   }
+
+  public boolean isSniff() {
+    return sniff;
+  }
 }

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/connection/ElasticDataPointerBuilder.java

@@ -6,6 +6,7 @@ public class ElasticDataPointerBuilder {
 
   private String clusterName = "elasticsearch";
   private ElasticAddress address;
+  private boolean sniff = true;
 
   private ElasticDataPointerBuilder() {
   }
@@ -20,9 +21,14 @@ public ElasticDataPointerBuilder setClusterName(String clusterName) {
     return this;
   }
 
+  public ElasticDataPointerBuilder setSniff(boolean sniff) {
+    this.sniff = sniff;
+    return this;
+  }
+
   public ElasticDataPointer build() {
     return new ElasticDataPointer(address.getHost(), clusterName, address.getIndexName(), address.getTypeName(),
-        address.getPort());
+        address.getPort(), sniff);
   }
 
   public static ElasticDataPointerBuilder builder() {

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/connection/ElasticSearchClientFactory.java

@@ -15,7 +15,7 @@ private ElasticSearchClientFactory() {
 
   public static Client createClient(ElasticDataPointer elasticDataPointer) {
     Settings settings = Settings.settingsBuilder()
-        .put("client.transport.sniff", true)
+        .put("client.transport.sniff", elasticDataPointer.isSniff())
         .put(ClusterName.SETTING, elasticDataPointer.getClusterName())
         .build();
     TransportClient client = TransportClient.builder().settings(settings).build();

src/test/java/pl/allegro/tech/search/elasticsearch/tools/reindex/ReindexCommandParserTest.java

@@ -1,24 +1,20 @@
 package pl.allegro.tech.search.elasticsearch.tools.reindex;
 
-import jdk.nashorn.internal.parser.JSONParser;
 import junitparams.JUnitParamsRunner;
 import junitparams.Parameters;
-import org.elasticsearch.index.mapper.ObjectMappers;
-import org.elasticsearch.index.mapper.object.ObjectMapper;
 import org.elasticsearch.search.sort.SortOrder;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointerAssert;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.query.PrefixSegment;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.query.RangeSegment;
 
 import java.util.Optional;
 
 import static pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointerAssert.assertThat;
+import static pl.allegro.tech.search.elasticsearch.tools.reindex.query.PrefixSegmentAssert.assertThat;
 import static pl.allegro.tech.search.elasticsearch.tools.reindex.query.QuerySegmentationAssert.assertThat;
 import static pl.allegro.tech.search.elasticsearch.tools.reindex.query.RangeSegmentAssert.assertThat;
-import static pl.allegro.tech.search.elasticsearch.tools.reindex.query.PrefixSegmentAssert.assertThat;
 
 @RunWith(JUnitParamsRunner.class)
 public class ReindexCommandParserTest {
@@ -217,6 +213,39 @@ public void parseCommandWithSortOrderDefaultASC() {
     Assert.assertEquals(SortOrder.ASC, commandParser.getSegmentation().getQuery().getSortOrder());
   }
 
+  @Test
+  public void parseDisableClusterSniffing() {
+    //given
+    ReindexCommandParser commandParser = new ReindexCommandParser();
+    //when
+    boolean result = commandParser.tryParse(createArgumentArray(
+        "-sc", "sourceClusterName",
+        "-tc", "targetClusterName",
+        "-s", "http://sourceHost1:9333/source_index/source_type",
+        "-t", "http://targetHost1:9333/target_index/target_type",
+        "-disable-cluster-sniffing"
+    ));
+    //then
+    Assert.assertFalse(commandParser.getSourcePointer().isSniff());
+    Assert.assertFalse(commandParser.getTargetPointer().isSniff());
+  }
+
+  @Test
+  public void parseWhenNotDisabledClusterSniffing() {
+    //given
+    ReindexCommandParser commandParser = new ReindexCommandParser();
+    //when
+    boolean result = commandParser.tryParse(createArgumentArray(
+        "-sc", "sourceClusterName",
+        "-tc", "targetClusterName",
+        "-s", "http://sourceHost1:9333/source_index/source_type",
+        "-t", "http://targetHost1:9333/target_index/target_type"
+    ));
+    //then
+    Assert.assertTrue(commandParser.getSourcePointer().isSniff());
+    Assert.assertTrue(commandParser.getTargetPointer().isSniff());
+
+  }
 
   private String[] createArgumentArray(String... args) {
     return args;

src/test/java/pl/allegro/tech/search/elasticsearch/tools/reindex/connection/ElasticSearchClientProducerTest.java

@@ -45,4 +45,35 @@ public void validateCreatedLocalElasticClientWithProperClusterName() throws Exce
     client.close();
   }
 
+
+  @Test
+  public void validateCreatedLocalElasticClientWithoutSniff() throws Exception {
+    //given
+    ElasticDataPointer dataPointer = ElasticDataPointerBuilder.builder()
+        .setAddress("http://localhost:9300/"+INDEX_NAME+"/type")
+        .setClusterName(CLUSTER_NAME)
+        .setSniff(false)
+        .build();
+    //when
+    Client client = ElasticSearchClientFactory.createClient(dataPointer);
+    //then
+    Assertions.assertThat(client.settings().get("cluster.name")).isEqualTo(CLUSTER_NAME);
+    client.close();
+  }
+
+  @Test
+  public void validateCreatedLocalElasticClientWithSniff() throws Exception {
+    //given
+    ElasticDataPointer dataPointer = ElasticDataPointerBuilder.builder()
+        .setAddress("http://localhost:9300/"+INDEX_NAME+"/type")
+        .setClusterName(CLUSTER_NAME)
+        .setSniff(true)
+        .build();
+    //when
+    Client client = ElasticSearchClientFactory.createClient(dataPointer);
+    //then
+    Assertions.assertThat(client.settings().get("cluster.name")).isEqualTo(CLUSTER_NAME);
+    client.close();
+  }
+
 }

src/test/java/pl/allegro/tech/search/elasticsearch/tools/reindex/embeded/EmbeddedElasticsearchCluster.java

@@ -11,7 +11,6 @@
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.node.NodeBuilder;
-import org.elasticsearch.search.sort.SortBuilder;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.ReindexInvokerTest;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointer;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointerBuilder;