natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Integrate pre-release Java Driver 2.2-rc1, custom build 

patch by Robert Stupp; reviewed by Ariel Weisberg for CASSANDRA-9493
This commit is contained in:
Robert Stupp 2015-06-04 11:17:05 +02:00
parent 49c31d6837
commit 16b02887df
11 changed files with 101 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

7
build.xml
View File

@ -386,7 +386,9 @@
<dependency groupId="io.netty" artifactId="netty-all" version="4.0.23.Final" />
<dependency groupId="com.google.code.findbugs" artifactId="jsr305" version="2.0.2" />
<dependency groupId="com.clearspring.analytics" artifactId="stream" version="2.5.2" />
<!-- TODO CASSANDRA-9543
<dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" version="2.1.5" classifier="shaded" />
-->
<dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj" version="4.4.2" />
<dependency groupId="org.caffinitas.ohc" artifactId="ohc-core" version="0.3.4" />
<dependency groupId="net.sf.supercsv" artifactId="super-csv" version="2.1.0" />
@ -438,7 +440,9 @@
<dependency groupId="org.apache.pig" artifactId="pig"/>
<dependency groupId="com.google.code.findbugs" artifactId="jsr305"/>
<dependency groupId="org.antlr" artifactId="antlr"/>
<!-- TODO CASSANDRA-9543
<dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" classifier="shaded"/>
-->
<dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj"/>
<dependency groupId="org.caffinitas.ohc" artifactId="ohc-core"/>
<dependency groupId="org.openjdk.jmh" artifactId="jmh-core"/>
@ -512,8 +516,9 @@
<dependency groupId="org.apache.hadoop" artifactId="hadoop-core" optional="true"/>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-minicluster" optional="true"/>
<dependency groupId="org.apache.pig" artifactId="pig" optional="true"/>
<!-- TODO CASSANDRA-9543
<dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" classifier="shaded" optional="true"/>
-->
<!-- don't need jna to run, but nice to have -->
<dependency groupId="net.java.dev.jna" artifactId="jna" version="4.0.0"/>

BIN
lib/cassandra-driver-core-2.1.5-shaded.jar
View File

Binary file not shown.

BIN
lib/cassandra-driver-core-2.2.0-rc1-SNAPSHOT-20150603-shaded.jar Normal file
View File

Binary file not shown.

22
src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java
View File

@ -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<Integer> getInputMinSimultReqPerConnections(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_MIN_SIMULT_REQ_PER_CONNECTION, conf);
}
public static Optional<Integer> 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<Integer> coreConnections = getInputCoreConnections(conf);
Optional<Integer> maxConnections = getInputMaxConnections(conf);
Optional<Integer> maxSimultaneousRequests = getInputMaxSimultReqPerConnections(conf);
Optional<Integer> 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");

69
src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java
View File

@ -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<Long, Row>
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<Long, Row>
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<Long, Row>
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<Long, Row>
}
@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

6
src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java
View File

@ -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)
{

5
src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
View File

@ -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"))

4
test/unit/org/apache/cassandra/cql3/CQLTester.java
View File

@ -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)
{

22
test/unit/org/apache/cassandra/cql3/UFTest.java
View File

@ -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);
}
}
}

BIN
tools/lib/cassandra-driver-core-2.0.9.2.jar
View File

Binary file not shown.