diff --git a/CHANGES.txt b/CHANGES.txt index f07592968d..933f5a6e1a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.2 + * Integrate pre-release Java Driver 2.2-rc1, custom build (CASSANDRA-9493) * Clean up gossiper logic for old versions (CASSANDRA-9370) * Fix custom payload coding/decoding to match the spec (CASSANDRA-9515) * ant test-all results incomplete when parsed (CASSANDRA-9463) diff --git a/build.xml b/build.xml index ecde9f9240..a548c177f1 100644 --- a/build.xml +++ b/build.xml @@ -386,7 +386,9 @@ + @@ -438,7 +440,9 @@ + @@ -512,8 +516,9 @@ + diff --git a/lib/cassandra-driver-core-2.1.5-shaded.jar b/lib/cassandra-driver-core-2.1.5-shaded.jar deleted file mode 100644 index bb83fb588e..0000000000 Binary files a/lib/cassandra-driver-core-2.1.5-shaded.jar and /dev/null differ diff --git a/lib/cassandra-driver-core-2.2.0-rc1-SNAPSHOT-20150603-shaded.jar b/lib/cassandra-driver-core-2.2.0-rc1-SNAPSHOT-20150603-shaded.jar new file mode 100644 index 0000000000..d75a595ac7 Binary files /dev/null and b/lib/cassandra-driver-core-2.2.0-rc1-SNAPSHOT-20150603-shaded.jar differ diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java index 9462724aa9..fe62ea763e 100644 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java +++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java @@ -41,6 +41,7 @@ import com.datastax.driver.core.AuthProvider; import com.datastax.driver.core.Cluster; import com.datastax.driver.core.HostDistance; import com.datastax.driver.core.PlainTextAuthProvider; +import com.datastax.driver.core.ProtocolVersion; import com.datastax.driver.core.policies.LoadBalancingPolicy; import com.datastax.driver.core.PoolingOptions; import com.datastax.driver.core.ProtocolOptions; @@ -48,7 +49,6 @@ import com.datastax.driver.core.QueryOptions; import com.datastax.driver.core.SSLOptions; import com.datastax.driver.core.SocketOptions; import org.apache.cassandra.hadoop.ConfigHelper; -import org.apache.cassandra.io.util.FileUtils; import org.apache.hadoop.conf.Configuration; @@ -65,7 +65,6 @@ public class CqlConfigHelper private static final String INPUT_NATIVE_PORT = "cassandra.input.native.port"; private static final String INPUT_NATIVE_CORE_CONNECTIONS_PER_HOST = "cassandra.input.native.core.connections.per.host"; private static final String INPUT_NATIVE_MAX_CONNECTIONS_PER_HOST = "cassandra.input.native.max.connections.per.host"; - private static final String INPUT_NATIVE_MIN_SIMULT_REQ_PER_CONNECTION = "cassandra.input.native.min.simult.reqs.per.connection"; private static final String INPUT_NATIVE_MAX_SIMULT_REQ_PER_CONNECTION = "cassandra.input.native.max.simult.reqs.per.connection"; private static final String INPUT_NATIVE_CONNECTION_TIMEOUT = "cassandra.input.native.connection.timeout"; private static final String INPUT_NATIVE_READ_CONNECTION_TIMEOUT = "cassandra.input.native.read.connection.timeout"; @@ -183,11 +182,6 @@ public class CqlConfigHelper return Integer.parseInt(conf.get(OUTPUT_NATIVE_PORT, "9042")); } - public static Optional getInputMinSimultReqPerConnections(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_MIN_SIMULT_REQ_PER_CONNECTION, conf); - } - public static Optional getInputMaxSimultReqPerConnections(Configuration conf) { return getIntSetting(INPUT_NATIVE_MAX_SIMULT_REQ_PER_CONNECTION, conf); @@ -336,7 +330,7 @@ public class CqlConfigHelper builder.withSSL(sslOptions.get()); if (protocolVersion.isPresent()) { - builder.withProtocolVersion(protocolVersion.get()); + builder.withProtocolVersion(ProtocolVersion.fromInt(protocolVersion.get())); } builder.withLoadBalancingPolicy(loadBalancingPolicy) .withSocketOptions(socketOptions) @@ -356,11 +350,6 @@ public class CqlConfigHelper conf.set(INPUT_NATIVE_MAX_CONNECTIONS_PER_HOST, connections); } - public static void setInputMinSimultReqPerConnections(Configuration conf, String reqs) - { - conf.set(INPUT_NATIVE_MIN_SIMULT_REQ_PER_CONNECTION, reqs); - } - public static void setInputMaxSimultReqPerConnections(Configuration conf, String reqs) { conf.set(INPUT_NATIVE_MAX_SIMULT_REQ_PER_CONNECTION, reqs); @@ -446,8 +435,7 @@ public class CqlConfigHelper Optional coreConnections = getInputCoreConnections(conf); Optional maxConnections = getInputMaxConnections(conf); Optional maxSimultaneousRequests = getInputMaxSimultReqPerConnections(conf); - Optional minSimultaneousRequests = getInputMinSimultReqPerConnections(conf); - + PoolingOptions poolingOptions = new PoolingOptions(); for (HostDistance hostDistance : Arrays.asList(HostDistance.LOCAL, HostDistance.REMOTE)) @@ -456,8 +444,6 @@ public class CqlConfigHelper poolingOptions.setCoreConnectionsPerHost(hostDistance, coreConnections.get()); if (maxConnections.isPresent()) poolingOptions.setMaxConnectionsPerHost(hostDistance, maxConnections.get()); - if (minSimultaneousRequests.isPresent()) - poolingOptions.setMinSimultaneousRequestsPerConnectionThreshold(hostDistance, minSimultaneousRequests.get()); if (maxSimultaneousRequests.isPresent()) poolingOptions.setMaxSimultaneousRequestsPerConnectionThreshold(hostDistance, maxSimultaneousRequests.get()); } @@ -602,7 +588,7 @@ public class CqlConfigHelper private static SSLContext getSSLContext(String truststorePath, String truststorePassword, String keystorePath, String keystorePassword) throws NoSuchAlgorithmException, KeyStoreException, CertificateException, IOException, UnrecoverableKeyException, KeyManagementException { - SSLContext ctx = null; + SSLContext ctx; try (FileInputStream tsf = new FileInputStream(truststorePath); FileInputStream ksf = new FileInputStream(keystorePath)) { ctx = SSLContext.getInstance("SSL"); diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java index 4a7bd595c3..3e79ad1679 100644 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java +++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java @@ -37,6 +37,7 @@ import org.slf4j.LoggerFactory; import com.datastax.driver.core.Cluster; import com.datastax.driver.core.ColumnDefinitions; import com.datastax.driver.core.ColumnMetadata; +import com.datastax.driver.core.DateWithoutTime; import com.datastax.driver.core.Metadata; import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row; @@ -138,7 +139,7 @@ public class CqlRecordReader extends RecordReader throw new RuntimeException("Can't create connection session"); //get negotiated serialization protocol - nativeProtocolVersion = cluster.getConfiguration().getProtocolOptions().getProtocolVersion(); + nativeProtocolVersion = cluster.getConfiguration().getProtocolOptions().getProtocolVersion().toInt(); // If the user provides a CQL query then we will use it without validation // otherwise we will fall back to building a query using the: @@ -332,6 +333,18 @@ public class CqlRecordReader extends RecordReader return row.isNull(name); } + @Override + public Object getObject(int i) + { + return row.getObject(i); + } + + @Override + public Object getObject(String s) + { + return row.getObject(s); + } + @Override public boolean getBool(int i) { @@ -344,6 +357,30 @@ public class CqlRecordReader extends RecordReader return row.getBool(name); } + @Override + public short getShort(int i) + { + return row.getShort(i); + } + + @Override + public short getShort(String s) + { + return row.getShort(s); + } + + @Override + public byte getByte(int i) + { + return row.getByte(i); + } + + @Override + public byte getByte(String s) + { + return row.getByte(s); + } + @Override public int getInt(int i) { @@ -369,15 +406,39 @@ public class CqlRecordReader extends RecordReader } @Override - public Date getDate(int i) + public Date getTimestamp(int i) + { + return row.getTimestamp(i); + } + + @Override + public Date getTimestamp(String s) + { + return row.getTimestamp(s); + } + + @Override + public DateWithoutTime getDate(int i) { return row.getDate(i); } @Override - public Date getDate(String name) + public DateWithoutTime getDate(String s) { - return row.getDate(name); + return row.getDate(s); + } + + @Override + public long getTime(int i) + { + return row.getTime(i); + } + + @Override + public long getTime(String s) + { + return row.getTime(s); } @Override diff --git a/src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java b/src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java index 87f96b35dd..da0822ac80 100644 --- a/src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java +++ b/src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java @@ -89,6 +89,12 @@ class LimitedLocalNodeFirstLocalBalancingPolicy implements LoadBalancingPolicy logger.debug("Initialized with replica hosts: {}", replicaHosts); } + @Override + public void close() + { + // + } + @Override public HostDistance distance(Host host) { diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java index 63baf9c4a8..537f30c401 100644 --- a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java +++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java @@ -95,7 +95,6 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo private String nativePort; private String nativeCoreConnections; private String nativeMaxConnections; - private String nativeMinSimultReqs; private String nativeMaxSimultReqs; private String nativeConnectionTimeout; private String nativeReadConnectionTimeout; @@ -577,8 +576,6 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo CqlConfigHelper.setInputCoreConnections(conf, nativeCoreConnections); if (nativeMaxConnections != null) CqlConfigHelper.setInputMaxConnections(conf, nativeMaxConnections); - if (nativeMinSimultReqs != null) - CqlConfigHelper.setInputMinSimultReqPerConnections(conf, nativeMinSimultReqs); if (nativeMaxSimultReqs != null) CqlConfigHelper.setInputMaxSimultReqPerConnections(conf, nativeMaxSimultReqs); if (nativeConnectionTimeout != null) @@ -796,8 +793,6 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo nativeCoreConnections = urlQuery.get("core_conns"); if (urlQuery.containsKey("max_conns")) nativeMaxConnections = urlQuery.get("max_conns"); - if (urlQuery.containsKey("min_simult_reqs")) - nativeMinSimultReqs = urlQuery.get("min_simult_reqs"); if (urlQuery.containsKey("max_simult_reqs")) nativeMaxSimultReqs = urlQuery.get("max_simult_reqs"); if (urlQuery.containsKey("native_timeout")) diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java index c37767c55d..d47b9d2a26 100644 --- a/test/unit/org/apache/cassandra/cql3/CQLTester.java +++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java @@ -82,11 +82,11 @@ public abstract class CQLTester static int maxProtocolVersion; static { int version; - for (version = 1; version <= Server.CURRENT_VERSION; version++) + for (version = 1; version <= Server.CURRENT_VERSION; ) { try { - ProtocolVersion.fromInt(version); + ProtocolVersion.fromInt(++version); } catch (IllegalArgumentException e) { diff --git a/test/unit/org/apache/cassandra/cql3/UFTest.java b/test/unit/org/apache/cassandra/cql3/UFTest.java index 9db50491f1..e1b2b5f928 100644 --- a/test/unit/org/apache/cassandra/cql3/UFTest.java +++ b/test/unit/org/apache/cassandra/cql3/UFTest.java @@ -25,6 +25,7 @@ import org.junit.Ignore; import org.junit.Test; import com.datastax.driver.core.*; +import com.datastax.driver.core.exceptions.InvalidQueryException; import org.apache.cassandra.cql3.functions.FunctionName; import org.apache.cassandra.cql3.functions.Functions; import org.apache.cassandra.cql3.functions.UDFunction; @@ -2139,7 +2140,6 @@ public class UFTest extends CQLTester } @Test - @Ignore("implement this unit test when Java Driver can handle new ExceptionCode.") public void testFunctionExecutionExceptionNet() throws Throwable { createTable("CREATE TABLE %s (key int primary key, dval double)"); @@ -2150,13 +2150,25 @@ public class UFTest extends CQLTester "RETURNS NULL ON NULL INPUT " + "RETURNS double " + "LANGUAGE JAVA\n" + - "AS 'throw new RuntimeException()';"); + "AS 'throw new RuntimeException();'"); for (int version = Server.VERSION_2; version <= maxProtocolVersion; version++) { - // TODO replace with appropiate code - assertRowsNet(version, - executeNet(version, "SELECT " + fName + "(dval) FROM %s WHERE key = 1")); + try + { + assertRowsNet(version, + executeNet(version, "SELECT " + fName + "(dval) FROM %s WHERE key = 1")); + Assert.fail(); + } + catch (com.datastax.driver.core.exceptions.FunctionExecutionException fee) + { + // Java driver neither throws FunctionExecutionException nor does it set the exception code correctly + Assert.assertTrue(version >= Server.VERSION_4); + } + catch (InvalidQueryException e) + { + Assert.assertTrue(version < Server.VERSION_4); + } } } diff --git a/tools/lib/cassandra-driver-core-2.0.9.2.jar b/tools/lib/cassandra-driver-core-2.0.9.2.jar deleted file mode 100644 index 3f82e770b0..0000000000 Binary files a/tools/lib/cassandra-driver-core-2.0.9.2.jar and /dev/null differ