Fix issues #33 and #35

- handle VARBINARY and LONGVARBINARY types with either ByteArrayInputStream
or byte[] in the methods CassandraPreparedStatement.setObject().
- fix configuration of the local datacenter using the one from the
configuration file when such a file is used.

Author: maximevw

Diff for: CHANGELOG.md

@@ -4,12 +4,21 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) and this project adheres to 
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [4.10.2] - 2023-11-01
+### Fixed
+- Fix issue [#33](https://github.com/ing-bank/cassandra-jdbc-wrapper/issues/33) to handle `VARBINARY` and 
+  `LONGVARBINARY` types with either `ByteArrayInputStream` or `byte[]` in the methods
+  `CassandraPreparedStatement.setObject()`.
+- Fix issue [#35](https://github.com/ing-bank/cassandra-jdbc-wrapper/issues/35) to fix configuration of the local
+  datacenter using the one from the configuration file when such a file is used.
+
 ## [4.10.1] - 2023-10-07
 ### Changed
 - Update Apache Commons IO to version 2.14.0.
 - Harmonize logging.
 ### Fixed
-- Fix multiple issues related to the method `findColumn(String)` of `CassandraResultSet` and `CassandraMetadataResultSet`:
+- Fix multiple issues related to the method `findColumn(String)` of `CassandraResultSet` and 
+  `CassandraMetadataResultSet`:
   - Fix issue [#31](https://github.com/ing-bank/cassandra-jdbc-wrapper/issues/31) to return a 1-based index value.
   - Return a result even if there's no row in the result set but the column exist in the statement.
   - Fix the exception thrown by the method when the given column name does not exist in the result set (was an
@@ -162,6 +171,7 @@ For this version, the changelog lists the main changes comparatively to the late
 - Fix logs in `CassandraConnection` constructor.
 
 [original project]: https://github.com/adejanovski/cassandra-jdbc-wrapper/
+[4.10.2]: https://github.com/ing-bank/cassandra-jdbc-wrapper/compare/v4.10.1...v4.10.2
 [4.10.1]: https://github.com/ing-bank/cassandra-jdbc-wrapper/compare/v4.10.0...v4.10.1
 [4.10.0]: https://github.com/ing-bank/cassandra-jdbc-wrapper/compare/v4.9.1...v4.10.0
 [4.9.1]: https://github.com/ing-bank/cassandra-jdbc-wrapper/compare/v4.9.0...v4.9.1

Diff for: pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>com.ing.data</groupId>
 	<artifactId>cassandra-jdbc-wrapper</artifactId>
-    <version>4.10.1</version>
+    <version>4.10.2</version>
 	<packaging>jar</packaging>
 
     <name>Cassandra JDBC Wrapper</name>

Diff for: src/main/java/com/ing/data/cassandra/jdbc/CassandraPreparedStatement.java

@@ -495,6 +495,8 @@ public void setObject(final int parameterIndex, final Object x) throws SQLExcept
             targetType = Types.BIGINT;
         } else if (x.getClass().equals(ByteArrayInputStream.class)) {
             targetType = Types.BINARY;
+        } else if (x instanceof byte[]) {
+            targetType = Types.BINARY;
         } else if (x.getClass().equals(String.class)) {
             targetType = Types.VARCHAR;
         } else if (x.getClass().equals(Boolean.class)) {
@@ -554,11 +556,20 @@ public final void setObject(final int parameterIndex, final Object x, final int
                 }
                 break;
             case Types.BINARY:
-                final byte[] array = new byte[((ByteArrayInputStream) x).available()];
-                try {
-                    ((ByteArrayInputStream) x).read(array);
-                } catch (final IOException e) {
-                    LOG.warn("Exception while setting object of BINARY type.", e);
+            case Types.VARBINARY:
+            case Types.LONGVARBINARY:
+                final byte[] array;
+                if (x instanceof ByteArrayInputStream) {
+                    array = new byte[((ByteArrayInputStream) x).available()];
+                    try {
+                        ((ByteArrayInputStream) x).read(array);
+                    } catch (final IOException e) {
+                        LOG.warn("Exception while setting object of BINARY/VARBINARY/LONGVARBINARY type.", e);
+                    }
+                } else if (x instanceof byte[]) {
+                    array = (byte[]) x;
+                } else {
+                    throw new SQLException("Unsupported parameter type: " + x.getClass());
                 }
                 this.boundStatement = this.boundStatement.setByteBuffer(parameterIndex - 1, ByteBuffer.wrap(array));
                 break;

Diff for: src/main/java/com/ing/data/cassandra/jdbc/SessionHolder.java

@@ -183,10 +183,12 @@ boolean acquire() {
 
     private Session createSession(final Properties properties) throws SQLException {
         File configurationFile = null;
+        boolean configurationFileExists = false;
         final String configurationFilePath = properties.getProperty(TAG_CONFIG_FILE, StringUtils.EMPTY);
         if (StringUtils.isNotBlank(configurationFilePath)) {
             configurationFile = new File(configurationFilePath);
-            if (configurationFile.exists()) {
+            configurationFileExists = configurationFile.exists();
+            if (configurationFileExists) {
                 // We remove some parameters to use the values defined into the specified configuration file
                 // instead.
                 this.properties.remove(TAG_CONSISTENCY_LEVEL);
@@ -264,7 +266,13 @@ private Session createSession(final Properties properties) throws SQLException {
         }
 
         // The DefaultLoadBalancingPolicy requires to specify a local data center.
-        builder.withLocalDatacenter(localDatacenter);
+        // Note (issue #35): This should only be set programmatically when there is no configuration file specified.
+        // When a configuration file is used, we rely on the property 'basic.load-balancing-policy.local-datacenter'
+        // of the configuration file, so we must not call withLocalDatacenter() method because when both are specified,
+        // the programmatic value takes precedence.
+        if (configurationFile == null || !configurationFileExists) {
+            builder.withLocalDatacenter(localDatacenter);
+        }
         if (loadBalancingPolicy.length() > 0) {
             // if a custom load balancing policy has been given in the JDBC URL, parse it and add it to the cluster
             // builder.

Diff for: src/test/java/com/ing/data/cassandra/jdbc/ConnectionUnitTest.java

@@ -41,6 +41,10 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.sql.Connection;
 import java.sql.DatabaseMetaData;
 import java.sql.ResultSet;
@@ -83,7 +87,8 @@ class ConnectionUnitTest extends UsingCassandraContainerTest {
 
     @Test
     void givenInvalidConfigurationFile_whenGetConnection_createConnectionIgnoringConfigFile() throws Exception {
-        initConnection(KEYSPACE, "configfile=wrong_application.conf", "consistency=LOCAL_QUORUM");
+        initConnection(KEYSPACE, "configfile=wrong_application.conf", "consistency=LOCAL_QUORUM",
+            "localdatacenter=datacenter1");
         assertNotNull(sqlConnection);
         assertNotNull(sqlConnection.getDefaultConsistencyLevel());
         final ConsistencyLevel consistencyLevel = sqlConnection.getDefaultConsistencyLevel();
@@ -377,6 +382,35 @@ void givenEnabledSslWithJsse_whenConfigureSsl_addDefaultSslEngineFactoryToSessio
         sqlConnection.close();
     }
 
+    @Test
+    void givenConfigurationFileWithSslEnabled_whenGetConnection_createConnectionWithExpectedConfig() throws Exception {
+        final ClassLoader classLoader = this.getClass().getClassLoader();
+        final URL confTestUrl = classLoader.getResource("test_application_with_ssl.conf");
+        if (confTestUrl == null) {
+            fail("Unable to find test_application_with_ssl.conf");
+        }
+
+        // Update the truststore path in the configuration file and store the modified file in a temporary location.
+        String content = new String(Files.readAllBytes(Paths.get(confTestUrl.toURI())));
+        content = content.replaceAll("\\$TRUSTSTORE_PATH",
+            Objects.requireNonNull(classLoader.getResource("cassandra.truststore")).getPath());
+        final Path updatedConfTestPath = Files.createTempFile("test_application_with_ssl_", ".conf");
+        Files.write(updatedConfTestPath, content.getBytes(StandardCharsets.UTF_8));
+
+        initConnection(KEYSPACE, "configfile=" + updatedConfTestPath);
+        assertNotNull(sqlConnection);
+        assertNotNull(sqlConnection.getSession());
+        assertNotNull(sqlConnection.getSession().getContext());
+        assertTrue(sqlConnection.getSession().getContext().getSslEngineFactory().isPresent());
+
+        final Statement statement = sqlConnection.createStatement();
+        final ResultSet resultSet = statement.executeQuery("SELECT * FROM system.local");
+        assertNotNull(resultSet);
+        resultSet.close();
+        statement.close();
+        sqlConnection.close();
+    }
+
     @Test
     void givenSslEngineFactory_whenConfigureSsl_addGivenSslEngineFactoryToSessionBuilder() throws Exception {
         final SessionHolder sessionHolder = new SessionHolder(Collections.singletonMap(URL_KEY,

Diff for: src/test/java/com/ing/data/cassandra/jdbc/JdbcRegressionUnitTest.java

@@ -984,4 +984,52 @@ void testIngIssue13_ResultSet() throws Exception {
                 "Cassandra results cannot be unwrapped to String");
         }
     }
+
+    @Test
+    @SuppressWarnings("ResultOfMethodCallIgnored")
+    void testIngIssue33() throws Exception {
+        final Statement stmt = sqlConnection.createStatement();
+
+        // Create the target table.
+        final String createTableQuery = "CREATE TABLE t33_blob (key int PRIMARY KEY, blob_col blob);";
+        stmt.execute(createTableQuery);
+        stmt.close();
+        sqlConnection.close();
+
+        // Open it up again to see the new column family.
+        sqlConnection = newConnection(KEYSPACE, "localdatacenter=datacenter1");
+        final String insertQuery = "INSERT INTO t33_blob (key, blob_col) VALUES(?, ?);";
+        final PreparedStatement stmt2 = sqlConnection.prepareStatement(insertQuery);
+        stmt2.setObject(1, 1);
+        stmt2.setObject(2, "test123".getBytes(StandardCharsets.UTF_8));
+        stmt2.execute();
+        stmt2.setObject(1, 2);
+        stmt2.setObject(2, "test456".getBytes(StandardCharsets.UTF_8), Types.VARBINARY);
+        stmt2.execute();
+        stmt2.setObject(1, 3);
+        stmt2.setObject(2, "test789".getBytes(StandardCharsets.UTF_8), Types.LONGVARBINARY);
+        stmt2.execute();
+
+        final Statement stmt3 = sqlConnection.createStatement();
+        ResultSet result = stmt3.executeQuery("SELECT * FROM t33_blob where key = 1;");
+        assertTrue(result.next());
+        byte[] array = new byte[result.getBinaryStream("blob_col").available()];
+        result.getBinaryStream("blob_col").read(array);
+        assertEquals("test123", new String(array, StandardCharsets.UTF_8));
+
+        result = stmt3.executeQuery("SELECT * FROM t33_blob where key = 2;");
+        assertTrue(result.next());
+        array = new byte[result.getBinaryStream("blob_col").available()];
+        result.getBinaryStream("blob_col").read(array);
+        assertEquals("test456", new String(array, StandardCharsets.UTF_8));
+
+        result = stmt3.executeQuery("SELECT * FROM t33_blob where key = 3;");
+        assertTrue(result.next());
+        array = new byte[result.getBinaryStream("blob_col").available()];
+        result.getBinaryStream("blob_col").read(array);
+        assertEquals("test789", new String(array, StandardCharsets.UTF_8));
+
+        stmt2.close();
+        stmt3.close();
+    }
 }

Diff for: src/test/resources/test_application_with_ssl.conf

@@ -0,0 +1,30 @@
+datastax-java-driver {
+  # The two following properties should be ignored by the JDBC wrapper.
+  basic.contact-points = [ "fake-server:9042" ]
+  basic.session-keyspace = testKeyspace
+
+  # All the following properties should be taken into account by the JDBC wrapper.
+  basic.request {
+    consistency = LOCAL_ONE
+    timeout = 10 seconds
+  }
+
+  basic.load-balancing-policy {
+    class = DefaultLoadBalancingPolicy
+    local-datacenter = datacenter1
+  }
+
+  advanced.auth-provider {
+    class = PlainTextAuthProvider
+    username = testUser
+    password = testPassword
+  }
+
+  advanced.ssl-engine-factory {
+    class = DefaultSslEngineFactory
+    hostname-validation = false
+    # The variable 'TRUSTSTORE_PATH' is replaced by the real truststore path in the tests using this configuration file.
+    truststore-path = $TRUSTSTORE_PATH
+    truststore-password = changeit
+  }
+}