Add support for multiple hosts with specific ports (#41)

Author: maximevw

CHANGELOG.md

@@ -6,6 +6,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ## Unreleased
 ### Added
+- Add support for connections with multiple contact points using different ports (see feature request 
+  [#41](https://github.com/ing-bank/cassandra-jdbc-wrapper/issues/41)).
 - Handle additional types and conversions in the methods `CassandraPreparedStatement.setObject()`:
   - JDBC types `BLOB`, `CLOB`, `NCLOB` and Java types `java.sql.Blob`, `java.sql.Clob`, and `java.sql.NClob` handled as
     arrays of bytes (CQL type `blob`)

README.md

@@ -114,15 +114,27 @@ Connect to a Cassandra cluster using the following arguments:
   please read the section "[Connecting to DBaaS](#connecting-to-dbaas)"; to use a configuration file, please read the 
   section "[Using a configuration file](#using-a-configuration-file)")
 
-You can give the driver any number of hosts you want separated by "--".
+You can give the driver any number of hosts you want separated by "--". You can optionally specify a port for each host.
+If only one port is specified after all the listed hosts, it applies to all hosts. If no port is specified at all, the
+default Cassandra port (9042) is used.
 They will be used as contact points for the driver to discover the entire cluster.
 Give enough hosts taking into account that some nodes may be unavailable upon establishing the JDBC connection.
 
+Here are some examples of connection strings with single or multiple contact points:
+
+| Valid JDBC URL                                               | Contact points used for connection |
+|--------------------------------------------------------------|------------------------------------|
+| jdbc:cassandra://localhost/keyspace                          | localhost:9042                     |
+| jdbc:cassandra://localhost:9043/keyspace                     | localhost:9043                     |
+| jdbc:cassandra://host1--host2/keyspace                       | host1:9042, host2:9042             |
+| jdbc:cassandra://host1--host2:9043/keyspace                  | host1:9043, host2:9043             |
+| jdbc:cassandra://host1:9042--host2--host3:9043/keyspace      | host1:9042, host2:9043, host3:9043 |
+
 You also have to specify the name of the local data center to use when the default load balancing policy is defined 
 (see paragraph below about load balancing policies) and no configuration file is specified. 
 
 Statements and prepared statements can be executed as with any JDBC driver, but note that queries must be expressed in 
-CQL3.
+CQL3 (Cassandra Query Language).
 
 Java example:
 ```java

src/main/java/com/ing/data/cassandra/jdbc/CassandraDataSource.java

@@ -17,30 +17,28 @@
 
 import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.datastax.oss.driver.internal.core.loadbalancing.DefaultLoadBalancingPolicy;
-import com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil;
+import com.ing.data.cassandra.jdbc.utils.ContactPoint;
 
 import javax.sql.ConnectionPoolDataSource;
 import javax.sql.DataSource;
 import java.io.PrintWriter;
 import java.sql.DriverManager;
 import java.sql.SQLException;
 import java.sql.SQLFeatureNotSupportedException;
-import java.sql.SQLNonTransientConnectionException;
+import java.util.Collections;
+import java.util.List;
 import java.util.Properties;
 import java.util.logging.Logger;
 
-import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.HOST_REQUIRED;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.NOT_SUPPORTED;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.NO_INTERFACE;
-import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.DEFAULT_PORT;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.PROTOCOL;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CONSISTENCY_LEVEL;
+import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CONTACT_POINTS;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CQL_VERSION;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_DATABASE_NAME;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_LOCAL_DATACENTER;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_PASSWORD;
-import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_PORT_NUMBER;
-import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_SERVER_NAME;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_USER;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.createSubName;
 
@@ -63,13 +61,9 @@ public class CassandraDataSource implements ConnectionPoolDataSource, DataSource
      */
     protected static final String DATA_SOURCE_DESCRIPTION = "Cassandra Data Source";
     /**
-     * The server host name where the data source is located.
+     * The contact points of the data source.
      */
-    protected String serverName;
-    /**
-     * The port number of the data source, by default {@value JdbcUrlUtil#DEFAULT_PORT}.
-     */
-    protected int portNumber = DEFAULT_PORT;
+    protected List<ContactPoint> contactPoints;
     /**
      * The database name. In case of Cassandra, i.e. the keyspace used as data source.
      */
@@ -110,10 +104,26 @@ public class CassandraDataSource implements ConnectionPoolDataSource, DataSource
      * @param user          The username used to connect.
      * @param password      The password used to connect.
      * @param consistency   The consistency level.
+     * @deprecated For removal. Use {@link #CassandraDataSource(List, String, String, String, String)} instead.
      */
+    @Deprecated
     public CassandraDataSource(final String host, final int port, final String keyspace, final String user,
                                final String password, final String consistency) {
-        this(host, port, keyspace, user, password, null, consistency, null);
+        this(Collections.singletonList(ContactPoint.of(host, port)), keyspace, user, password, null, consistency, null);
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param contactPoints The contact points.
+     * @param keyspace      The keyspace.
+     * @param user          The username used to connect.
+     * @param password      The password used to connect.
+     * @param consistency   The consistency level.
+     */
+    public CassandraDataSource(final List<ContactPoint> contactPoints, final String keyspace, final String user,
+                               final String password, final String consistency) {
+        this(contactPoints, keyspace, user, password, null, consistency, null);
     }
 
     /**
@@ -126,34 +136,31 @@ public CassandraDataSource(final String host, final int port, final String keysp
      * @param password      The password used to connect.
      * @param version       The CQL version. Deprecated, do not use anymore.
      * @param consistency   The consistency level.
-     * @deprecated For removal. Use {@link #CassandraDataSource(String, int, String, String, String, String)} instead.
+     * @deprecated For removal. Use {@link #CassandraDataSource(List, String, String, String, String)} instead.
      */
     @Deprecated
     public CassandraDataSource(final String host, final int port, final String keyspace, final String user,
                                final String password, final String version, final String consistency) {
-        this(host, port, keyspace, user, password, version, consistency, null);
+        this(Collections.singletonList(ContactPoint.of(host, port)),
+            keyspace, user, password, version, consistency, null);
     }
 
     /**
      * Constructor specifying a local datacenter (required to use {@link DefaultLoadBalancingPolicy}).
      *
-     * @param host              The host name.
-     * @param port              The port.
+     * @param contactPoints     The contact points.
      * @param keyspace          The keyspace.
      * @param user              The username used to connect.
      * @param password          The password used to connect.
      * @param version           The CQL version. Deprecated, do not use anymore.
      * @param consistency       The consistency level.
      * @param localDataCenter   The local datacenter.
      */
-    public CassandraDataSource(final String host, final int port, final String keyspace, final String user,
+    public CassandraDataSource(final List<ContactPoint> contactPoints, final String keyspace, final String user,
                                final String password, final String version, final String consistency,
                                final String localDataCenter) {
-        if (host != null) {
-            setServerName(host);
-        }
-        if (port >= 0) {
-            setPortNumber(port);
+        if (contactPoints != null && !contactPoints.isEmpty()) {
+            setContactPoints(contactPoints);
         }
         if (version != null) {
             setVersion(version);
@@ -179,21 +186,21 @@ public String getDescription() {
     }
 
     /**
-     * Gets the server host name where the data source is located.
+     * Gets the contact points of the data source.
      *
-     * @return The server host name where the data source is located.
+     * @return The contact points of the data source.
      */
-    public String getServerName() {
-        return this.serverName;
+    public List<ContactPoint> getContactPoints() {
+        return this.contactPoints;
     }
 
     /**
-     * Sets the server host name where the data source is located.
+     * Sets the contact points of the data source.
      *
-     * @param serverName The host name.
+     * @param contactPoints The contact points of the data source.
      */
-    public void setServerName(final String serverName) {
-        this.serverName = serverName;
+    public void setContactPoints(final List<ContactPoint> contactPoints) {
+        this.contactPoints = contactPoints;
     }
 
     /**
@@ -246,24 +253,6 @@ public void setConsistency(final String consistency) {
         this.consistency = consistency;
     }
 
-    /**
-     * Gets the port number of the data source.
-     *
-     * @return The port number of the data source.
-     */
-    public int getPortNumber() {
-        return this.portNumber;
-    }
-
-    /**
-     * Sets the port number of the data source.
-     *
-     * @param portNumber The port number of the data source.
-     */
-    public void setPortNumber(final int portNumber) {
-        this.portNumber = portNumber;
-    }
-
     /**
      * Gets the database name. In case of Cassandra, i.e. the keyspace used as data source.
      *
@@ -347,12 +336,9 @@ public CassandraConnection getConnection(final String user, final String passwor
         this.user = user;
         this.password = password;
 
-        if (this.serverName != null) {
-            props.setProperty(TAG_SERVER_NAME, this.serverName);
-        } else {
-            throw new SQLNonTransientConnectionException(HOST_REQUIRED);
+        if (this.contactPoints != null && !this.contactPoints.isEmpty()) {
+            props.put(TAG_CONTACT_POINTS, this.contactPoints);
         }
-        props.setProperty(TAG_PORT_NUMBER, String.valueOf(this.portNumber));
         if (this.databaseName != null) {
             props.setProperty(TAG_DATABASE_NAME, this.databaseName);
         }

src/main/java/com/ing/data/cassandra/jdbc/CassandraDriver.java

@@ -36,6 +36,7 @@
 import static com.ing.data.cassandra.jdbc.utils.DriverUtil.parseVersion;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.NOT_SUPPORTED;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.PROTOCOL;
+import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CONTACT_POINTS;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_PASSWORD;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_USER;
 
@@ -75,7 +76,9 @@ public Connection connect(final String url, final Properties properties) throws
             final Enumeration<Object> keys = properties.keys();
             while (keys.hasMoreElements()) {
                 final String key = (String) keys.nextElement();
-                params.put(key, properties.getProperty(key));
+                if (!TAG_CONTACT_POINTS.equals(key)) {
+                    params.put(key, properties.getProperty(key));
+                }
             }
             params.put(SessionHolder.URL_KEY, url);
 

src/main/java/com/ing/data/cassandra/jdbc/SessionHolder.java

@@ -40,19 +40,18 @@
 import com.ing.data.cassandra.jdbc.codec.TimestampToLongCodec;
 import com.ing.data.cassandra.jdbc.codec.TinyintToIntCodec;
 import com.ing.data.cassandra.jdbc.codec.VarintToIntCodec;
+import com.ing.data.cassandra.jdbc.utils.ContactPoint;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.math.NumberUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
-import java.net.InetSocketAddress;
 import java.sql.SQLException;
 import java.sql.SQLNonTransientConnectionException;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -70,18 +69,17 @@
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CONFIG_FILE;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CONNECT_TIMEOUT;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CONSISTENCY_LEVEL;
+import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CONTACT_POINTS;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_DATABASE_NAME;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_DEBUG;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_ENABLE_SSL;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_KEEP_ALIVE;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_LOAD_BALANCING_POLICY;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_LOCAL_DATACENTER;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_PASSWORD;
-import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_PORT_NUMBER;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_RECONNECT_POLICY;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_REQUEST_TIMEOUT;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_RETRY_POLICY;
-import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_SERVER_NAME;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_SSL_ENGINE_FACTORY;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_SSL_HOSTNAME_VERIFICATION;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_TCP_NO_DELAY;
@@ -181,6 +179,7 @@ boolean acquire() {
         }
     }
 
+    @SuppressWarnings("unchecked")
     private Session createSession(final Properties properties) throws SQLException {
         File configurationFile = null;
         boolean configurationFileExists = false;
@@ -209,8 +208,7 @@ private Session createSession(final Properties properties) throws SQLException {
             }
         }
 
-        final String hosts = properties.getProperty(TAG_SERVER_NAME);
-        final int port = Integer.parseInt(properties.getProperty(TAG_PORT_NUMBER));
+        final List<ContactPoint> contactPoints = (List<ContactPoint>) properties.get(TAG_CONTACT_POINTS);
         final String cloudSecureConnectBundle = properties.getProperty(TAG_CLOUD_SECURE_CONNECT_BUNDLE);
         final String keyspace = properties.getProperty(TAG_DATABASE_NAME);
         final String username = properties.getProperty(TAG_USER, StringUtils.EMPTY);
@@ -243,12 +241,14 @@ private Session createSession(final Properties properties) throws SQLException {
         if (StringUtils.isNotBlank(cloudSecureConnectBundle)) {
             driverConfigLoaderBuilder.withString(DefaultDriverOption.CLOUD_SECURE_CONNECT_BUNDLE,
                 cloudSecureConnectBundle);
-            LOG.info("Cloud secure connect bundle used. Host(s) {} will be ignored.", hosts);
+            LOG.info("Cloud secure connect bundle used. Host(s) {} will be ignored.",
+                    contactPoints.stream()
+                        .map(ContactPoint::toString)
+                        .collect(Collectors.joining(", ")));
         } else {
-            builder.addContactPoints(Arrays.stream(hosts.split("--"))
-	            .map(host -> InetSocketAddress.createUnresolved(host, port))
-	            .collect(Collectors.toList())
-            );
+            builder.addContactPoints(contactPoints.stream()
+                .map(ContactPoint::toInetSocketAddress)
+                .collect(Collectors.toList()));
         }
 
         // Set request timeout (in milliseconds) if defined.