DATAES-470 - Fixed parsing of cluster nodes in TransportClientFactoryBean.

Extracted ClusterNodes value object to capture the parsing logic and actually properly test it. Added unit tests to verify the proper rejection and the two cases outlined in the ticket.

Related tickets: DATAES-283.

Author: odrotbohm

src/main/java/org/springframework/data/elasticsearch/client/ClusterNodes.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * 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.
+ */
+package org.springframework.data.elasticsearch.client;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.springframework.data.util.Streamable;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Value object to represent a list of cluster nodes.
+ *
+ * @author Oliver Gierke
+ * @since 3.0.9
+ */
+class ClusterNodes implements Streamable<TransportAddress> {
+
+	public static ClusterNodes DEFAULT = ClusterNodes.of("127.0.0.1:9300");
+
+	private static final String COLON = ":";
+	private static final String COMMA = ",";
+
+	private final List<TransportAddress> clusterNodes;
+
+	/**
+	 * Creates a new {@link ClusterNodes} by parsing the given source.
+	 *
+	 * @param source must not be {@literal null} or empty.
+	 */
+	private ClusterNodes(String source) {
+
+		Assert.hasText(source, "Cluster nodes source must not be null or empty!");
+
+		String[] nodes = StringUtils.delimitedListToStringArray(source, COMMA);
+
+		this.clusterNodes = Arrays.stream(nodes).map(node -> {
+
+			String[] segments = StringUtils.delimitedListToStringArray(node, COLON);
+
+			Assert.isTrue(segments.length == 2,
+					() -> String.format("Invalid cluster node %s in %s! Must be in the format host:port!", node, source));
+
+			String host = segments[0].trim();
+			String port = segments[1].trim();
+
+			Assert.hasText(host, () -> String.format("No host name given cluster node %s!", node));
+			Assert.hasText(port, () -> String.format("No port given in cluster node %s!", node));
+
+			return new InetSocketTransportAddress(toInetAddress(host), Integer.valueOf(port));
+
+		}).collect(Collectors.toList());
+	}
+
+	/**
+	 * Creates a new {@link ClusterNodes} by parsing the given source. The expected format is a comma separated list of
+	 * host-port-combinations separated by a colon: {@code host:port,host:port,…}.
+	 *
+	 * @param source must not be {@literal null} or empty.
+	 * @return
+	 */
+	public static ClusterNodes of(String source) {
+		return new ClusterNodes(source);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Iterable#iterator()
+	 */
+	@Override
+	public Iterator<TransportAddress> iterator() {
+		return clusterNodes.iterator();
+	}
+
+	private static InetAddress toInetAddress(String host) {
+
+		try {
+			return InetAddress.getByName(host);
+		} catch (UnknownHostException o_O) {
+			throw new IllegalArgumentException(o_O);
+		}
+	}
+}

src/main/java/org/springframework/data/elasticsearch/client/TransportClientFactoryBean.java

@@ -18,7 +18,6 @@
 import io.netty.util.ThreadDeathWatcher;
 import io.netty.util.concurrent.GlobalEventExecutor;
 
-import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -31,7 +30,6 @@
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.network.NetworkModule;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.transport.InetSocketTransportAddress;
 import org.elasticsearch.index.reindex.ReindexPlugin;
 import org.elasticsearch.join.ParentJoinPlugin;
 import org.elasticsearch.percolator.PercolatorPlugin;
@@ -46,7 +44,6 @@
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
-import org.springframework.util.StringUtils;
 
 /**
  * TransportClientFactoryBean
@@ -55,21 +52,19 @@
  * @author Mohsin Husen
  * @author Jakub Vavrik
  * @author Piotr Betkier
+ * @author Oliver Gierke
  */
-
 public class TransportClientFactoryBean implements FactoryBean<TransportClient>, InitializingBean, DisposableBean {
 
 	private static final Logger logger = LoggerFactory.getLogger(TransportClientFactoryBean.class);
-	private String clusterNodes = "127.0.0.1:9300";
+	private ClusterNodes clusterNodes = ClusterNodes.of("127.0.0.1:9300");
 	private String clusterName = "elasticsearch";
 	private Boolean clientTransportSniff = true;
 	private Boolean clientIgnoreClusterName = Boolean.FALSE;
 	private String clientPingTimeout = "5s";
 	private String clientNodesSamplerInterval = "5s";
 	private TransportClient client;
 	private Properties properties;
-	static final String COLON = ":";
-	static final String COMMA = ",";
 
 	@Override
 	public void destroy() throws Exception {
@@ -106,39 +101,26 @@ public void afterPropertiesSet() throws Exception {
 	protected void buildClient() throws Exception {
 
 		client = new SpringDataTransportClient(settings());
-		Assert.hasText(clusterNodes, "[Assertion failed] clusterNodes settings missing.");
-		String[] clusterNodesArray = StringUtils.split(clusterNodes, COMMA);
-		if (clusterNodesArray != null) {
-			for (String clusterNode : clusterNodesArray) {
-				if (clusterNode != null) {
-					int colonPosition = clusterName.lastIndexOf(COLON);
-					String hostName = colonPosition != -1 ? clusterNode.substring(0, colonPosition) : clusterNode;
-					String port = colonPosition != -1 ? clusterNode.substring(colonPosition, clusterNode.length()) : "";
-					Assert.hasText(hostName, "[Assertion failed] missing host name in 'clusterNodes'");
-					Assert.hasText(port, "[Assertion failed] missing port in 'clusterNodes'");
-					logger.info("adding transport node : " + clusterNode);
-					client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(hostName), Integer.valueOf(port)));
-				}
-			}
-		}
+
+		clusterNodes.stream() //
+				.peek(it -> logger.info("Adding transport node : " + it.toString())) //
+				.forEach(client::addTransportAddress);
+
 		client.connectedNodes();
 	}
 
 	private Settings settings() {
 		if (properties != null) {
 			return Settings.builder().put(properties).build();
 		}
-		return Settings.builder()
-				.put("cluster.name", clusterName)
-				.put("client.transport.sniff", clientTransportSniff)
+		return Settings.builder().put("cluster.name", clusterName).put("client.transport.sniff", clientTransportSniff)
 				.put("client.transport.ignore_cluster_name", clientIgnoreClusterName)
 				.put("client.transport.ping_timeout", clientPingTimeout)
-				.put("client.transport.nodes_sampler_interval", clientNodesSamplerInterval)
-				.build();
+				.put("client.transport.nodes_sampler_interval", clientNodesSamplerInterval).build();
 	}
 
 	public void setClusterNodes(String clusterNodes) {
-		this.clusterNodes = clusterNodes;
+		this.clusterNodes = ClusterNodes.of(clusterNodes);
 	}
 
 	public void setClusterName(String clusterName) {
@@ -176,7 +158,7 @@ public void setClientIgnoreClusterName(Boolean clientIgnoreClusterName) {
 	public void setProperties(Properties properties) {
 		this.properties = properties;
 	}
-	
+
 	/**
 	 * Pretty exact copy of {@link PreBuiltTransportClient} except that we're inspecting the classpath for Netty
 	 * dependencies to only include the ones available. {@link PreBuiltTransportClient} expects both Netty 3 and Netty 4
@@ -236,8 +218,9 @@ private static void setSystemPropertyIfUnset(final String key, final String valu
 					found = true;
 				} catch (ClassNotFoundException | LinkageError e) {}
 			}
-			
-			Assert.state(found, "Neither Netty 3 or Netty 4 plugin found on the classpath. One of them is required to run the transport client!");
+
+			Assert.state(found,
+					"Neither Netty 3 or Netty 4 plugin found on the classpath. One of them is required to run the transport client!");
 
 			plugins.add(ReindexPlugin.class);
 			plugins.add(PercolatorPlugin.class);

src/test/java/org/springframework/data/elasticsearch/client/ClusterNodesUnitTests.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * 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.
+ */
+package org.springframework.data.elasticsearch.client;
+
+import static org.assertj.core.api.Assertions.*;
+
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link ClusterNodes}.
+ *
+ * @author Oliver Gierke
+ */
+public class ClusterNodesUnitTests {
+
+	@Test // DATAES-470
+	public void parsesSingleClusterNode() {
+
+		ClusterNodes nodes = ClusterNodes.DEFAULT;
+
+		assertThat(nodes).hasSize(1) //
+				.first().satisfies(it -> {
+					assertThat(it.getAddress()).isEqualTo("127.0.0.1");
+					assertThat(it.getPort()).isEqualTo(9300);
+				});
+	}
+
+	@Test // DATAES-470
+	public void parsesMultiClusterNode() {
+
+		ClusterNodes nodes = ClusterNodes.of("127.0.0.1:1234,10.1.0.1:5678");
+
+		assertThat(nodes.stream()).hasSize(2); //
+		assertThat(nodes.stream()).element(0).satisfies(it -> {
+			assertThat(it.getAddress()).isEqualTo("127.0.0.1");
+			assertThat(it.getPort()).isEqualTo(1234);
+		});
+		assertThat(nodes.stream()).element(1).satisfies(it -> {
+			assertThat(it.getAddress()).isEqualTo("10.1.0.1");
+			assertThat(it.getPort()).isEqualTo(5678);
+		});
+	}
+
+	@Test // DATAES-470
+	public void rejectsEmptyHostName() {
+
+		assertThatExceptionOfType(IllegalArgumentException.class) //
+				.isThrownBy(() -> ClusterNodes.of(":8080")) //
+				.withMessageContaining("host");
+	}
+
+	@Test // DATAES-470
+	public void rejectsEmptyPort() {
+
+		assertThatExceptionOfType(IllegalArgumentException.class) //
+				.isThrownBy(() -> ClusterNodes.of("localhost:")) //
+				.withMessageContaining("port");
+	}
+
+	@Test // DATAES-470
+	public void rejectsMissingPort() {
+
+		assertThatExceptionOfType(IllegalArgumentException.class) //
+				.isThrownBy(() -> ClusterNodes.of("localhost")) //
+				.withMessageContaining("host:port");
+	}
+
+	@Test // DATAES-470
+	public void rejectsUnresolvableHost() {
+
+		assertThatExceptionOfType(IllegalArgumentException.class) //
+				.isThrownBy(() -> ClusterNodes.of("mylocalhost:80"));
+	}
+}