You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2020/07/23 11:58:36 UTC
[camel] branch master updated: CAMEL-15293 Cassandraql: Upgrade
datastax driver to 4.7.2 (from 3.7.2) (#4000)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 4e44db6 CAMEL-15293 Cassandraql: Upgrade datastax driver to 4.7.2 (from 3.7.2) (#4000)
4e44db6 is described below
commit 4e44db64c1dc13d36e4ce71c9b1a78c60dc302f3
Author: JiriOndrusek <on...@gmail.com>
AuthorDate: Thu Jul 23 13:58:08 2020 +0200
CAMEL-15293 Cassandraql: Upgrade datastax driver to 4.7.2 (from 3.7.2) (#4000)
---
camel-dependencies/pom.xml | 8 +-
.../org/apache/camel/catalog/components/cql.json | 8 +-
.../apache/camel/catalog/docs/cql-component.adoc | 8 +-
components/camel-cassandraql/pom.xml | 8 +-
.../cassandra/CassandraEndpointConfigurer.java | 24 +--
.../org/apache/camel/component/cassandra/cql.json | 8 +-
.../src/main/docs/cql-component.adoc | 8 +-
.../component/cassandra/CassandraConsumer.java | 8 +-
.../component/cassandra/CassandraEndpoint.java | 133 ++++++++--------
.../component/cassandra/CassandraProducer.java | 30 ++--
.../cassandra/ResultSetConversionStrategies.java | 4 +-
.../cassandra/ResultSetConversionStrategy.java | 3 +-
.../cassandra/CassandraAggregationRepository.java | 63 ++++----
.../NamedCassandraAggregationRepository.java | 12 +-
.../cassandra/CassandraIdempotentRepository.java | 57 ++++---
.../NamedCassandraIdempotentRepository.java | 12 +-
.../cassandra/CassandraLoadBalancingPolicies.java | 57 -------
.../utils/cassandra/CassandraSessionHolder.java | 37 +----
.../camel/utils/cassandra/CassandraUtils.java | 74 +++++----
.../component/cassandra/BaseCassandraTest.java | 1 -
.../component/cassandra/CassandraCQLUnit.java | 26 +--
.../cassandra/CassandraComponentBeanRefTest.java | 17 --
.../cassandra/CassandraComponentConsumerTest.java | 2 +-
.../cassandra/CassandraComponentProducerTest.java | 80 +++++-----
.../CassandraComponentProducerUnpreparedTest.java | 42 +++--
.../component/cassandra/CassandraUnitUtils.java | 8 +-
.../cassandra/MockLoadBalancingPolicy.java | 44 +++++
.../ResultSetConversionStrategiesTest.java | 4 +-
.../CassandraAggregationRepositoryTest.java | 14 +-
.../CassandraAggregationSerializedHeadersTest.java | 2 +-
.../cassandra/CassandraAggregationTest.java | 2 +-
.../NamedCassandraAggregationRepositoryTest.java | 15 +-
.../CassandraIdempotentRepositoryTest.java | 5 +-
.../cassandra/CassandraIdempotentTest.java | 2 +-
.../NamedCassandraIdempotentRepositoryTest.java | 4 +-
.../src/test/resources/camel-cassandra.yaml | 3 +
.../builder/endpoint/StaticEndpointBuilders.java | 4 +-
.../dsl/CassandraEndpointBuilderFactory.java | 177 +++++++++------------
.../modules/ROOT/pages/cql-component.adoc | 8 +-
.../ROOT/pages/camel-3x-upgrade-guide-3_5.adoc | 14 ++
parent/pom.xml | 17 +-
41 files changed, 499 insertions(+), 554 deletions(-)
diff --git a/camel-dependencies/pom.xml b/camel-dependencies/pom.xml
index 64c70e7..a8ab467 100644
--- a/camel-dependencies/pom.xml
+++ b/camel-dependencies/pom.xml
@@ -112,10 +112,10 @@
<camel.osgi.symbolic.name>${project.groupId}.${project.artifactId}</camel.osgi.symbolic.name>
<camel.osgi.version>version=${project.version}</camel.osgi.version>
<camel.surefire.fork.vmargs></camel.surefire.fork.vmargs>
- <cassandra-driver-guava-version>19.0</cassandra-driver-guava-version>
- <cassandra-driver-version>3.7.2</cassandra-driver-version>
- <cassandra-unit-version>3.1.3.2</cassandra-unit-version>
- <cassandra-version>3.7</cassandra-version>
+ <cassandra-driver-guava-version>25.1-jre</cassandra-driver-guava-version>
+ <cassandra-driver-version>4.7.2</cassandra-driver-version>
+ <cassandra-unit-version>4.3.1.0</cassandra-unit-version>
+ <cassandra-version>3.11.5</cassandra-version>
<cdi-api-2.0-version>2.0</cdi-api-2.0-version>
<cglib-version>3.2.12</cglib-version>
<chunk-templates-version>3.5.0</chunk-templates-version>
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cql.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cql.json
index 78315b5..44c9be1 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cql.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/cql.json
@@ -30,15 +30,15 @@
"hosts": { "kind": "path", "displayName": "Hosts", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Hostname(s) cassansdra server(s). Multiple hosts can be separated by comma." },
"port": { "kind": "path", "displayName": "Port", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "description": "Port number of cassansdra server(s)" },
"keyspace": { "kind": "path", "displayName": "Keyspace", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Keyspace to use" },
- "cluster": { "kind": "parameter", "displayName": "Cluster", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.driver.core.Cluster", "deprecated": false, "secret": false, "description": "To use the Cluster instance (you would normally not use this option)" },
"clusterName": { "kind": "parameter", "displayName": "Cluster Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Cluster name" },
- "consistencyLevel": { "kind": "parameter", "displayName": "Consistency Level", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.driver.core.ConsistencyLevel", "enum": [ "ANY", "ONE", "TWO", "THREE", "QUORUM", "ALL", "LOCAL_QUORUM", "EACH_QUORUM", "SERIAL", "LOCAL_SERIAL", "LOCAL_ONE" ], "deprecated": false, "secret": false, "description": "Consistency level to use" },
+ "consistencyLevel": { "kind": "parameter", "displayName": "Consistency Level", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.oss.driver.api.core.DefaultConsistencyLevel", "enum": [ "ANY", "ONE", "TWO", "THREE", "QUORUM", "ALL", "LOCAL_ONE", "LOCAL_QUORUM", "EACH_QUORUM", "SERIAL", "LOCAL_SERIAL" ], "deprecated": false, "secret": false, "description": "Consistency level to use" },
"cql": { "kind": "parameter", "displayName": "Cql", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "CQL query to perform. Can be overridden with the message header with key CamelCqlQuery." },
- "loadBalancingPolicy": { "kind": "parameter", "displayName": "Load Balancing Policy", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "To use a specific LoadBalancingPolicy" },
+ "datacenter": { "kind": "parameter", "displayName": "Datacenter", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "datacenter1", "description": "Datacenter to use" },
+ "loadBalancingPolicyClass": { "kind": "parameter", "displayName": "Load Balancing Policy Class", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "To use a specific LoadBalancingPolicyClass" },
"password": { "kind": "parameter", "displayName": "Password", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Password for session authentication" },
"prepareStatements": { "kind": "parameter", "displayName": "Prepare Statements", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "description": "Whether to use PreparedStatements or regular Statements" },
"resultSetConversionStrategy": { "kind": "parameter", "displayName": "Result Set Conversion Strategy", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.cassandra.ResultSetConversionStrategy", "deprecated": false, "secret": false, "description": "To use a custom class that implements logic for converting ResultSet into message body ALL, ONE, LIMIT_10, LIMIT_100..." },
- "session": { "kind": "parameter", "displayName": "Session", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.driver.core.Session", "deprecated": false, "secret": false, "description": "To use the Session instance (you would normally not use this option)" },
+ "session": { "kind": "parameter", "displayName": "Session", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.oss.driver.api.core.CqlSession", "deprecated": false, "secret": false, "description": "To use the Session instance (you would normally not use this option)" },
"username": { "kind": "parameter", "displayName": "Username", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Username for session authentication" },
"bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled b [...]
"sendEmptyMessageWhenIdle": { "kind": "parameter", "displayName": "Send Empty Message When Idle", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead." },
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/cql-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/cql-component.adoc
index 5c58bbc..c8f242a 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/cql-component.adoc
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/cql-component.adoc
@@ -106,15 +106,15 @@ with the following path and query parameters:
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
-| *cluster* (common) | To use the Cluster instance (you would normally not use this option) | | Cluster
| *clusterName* (common) | Cluster name | | String
-| *consistencyLevel* (common) | Consistency level to use. The value can be one of: ANY, ONE, TWO, THREE, QUORUM, ALL, LOCAL_QUORUM, EACH_QUORUM, SERIAL, LOCAL_SERIAL, LOCAL_ONE | | ConsistencyLevel
+| *consistencyLevel* (common) | Consistency level to use. The value can be one of: ANY, ONE, TWO, THREE, QUORUM, ALL, LOCAL_ONE, LOCAL_QUORUM, EACH_QUORUM, SERIAL, LOCAL_SERIAL | | DefaultConsistencyLevel
| *cql* (common) | CQL query to perform. Can be overridden with the message header with key CamelCqlQuery. | | String
-| *loadBalancingPolicy* (common) | To use a specific LoadBalancingPolicy | | String
+| *datacenter* (common) | Datacenter to use | datacenter1 | String
+| *loadBalancingPolicyClass* (common) | To use a specific LoadBalancingPolicyClass | | String
| *password* (common) | Password for session authentication | | String
| *prepareStatements* (common) | Whether to use PreparedStatements or regular Statements | true | boolean
| *resultSetConversionStrategy* (common) | To use a custom class that implements logic for converting ResultSet into message body ALL, ONE, LIMIT_10, LIMIT_100... | | ResultSetConversionStrategy
-| *session* (common) | To use the Session instance (you would normally not use this option) | | Session
+| *session* (common) | To use the Session instance (you would normally not use this option) | | CqlSession
| *username* (common) | Username for session authentication | | String
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *sendEmptyMessageWhenIdle* (consumer) | If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead. | false | boolean
diff --git a/components/camel-cassandraql/pom.xml b/components/camel-cassandraql/pom.xml
index 4089621..dfe7bf0 100644
--- a/components/camel-cassandraql/pom.xml
+++ b/components/camel-cassandraql/pom.xml
@@ -45,8 +45,12 @@
<!-- cassandra -->
<dependency>
- <groupId>com.datastax.cassandra</groupId>
- <artifactId>cassandra-driver-core</artifactId>
+ <groupId>com.datastax.oss</groupId>
+ <artifactId>java-driver-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.datastax.oss</groupId>
+ <artifactId>java-driver-query-builder</artifactId>
</dependency>
<!-- logging -->
diff --git a/components/camel-cassandraql/src/generated/java/org/apache/camel/component/cassandra/CassandraEndpointConfigurer.java b/components/camel-cassandraql/src/generated/java/org/apache/camel/component/cassandra/CassandraEndpointConfigurer.java
index b0b1383..7beeb7f 100644
--- a/components/camel-cassandraql/src/generated/java/org/apache/camel/component/cassandra/CassandraEndpointConfigurer.java
+++ b/components/camel-cassandraql/src/generated/java/org/apache/camel/component/cassandra/CassandraEndpointConfigurer.java
@@ -29,12 +29,12 @@ public class CassandraEndpointConfigurer extends PropertyConfigurerSupport imple
case "basicPropertyBinding": target.setBasicPropertyBinding(property(camelContext, boolean.class, value)); return true;
case "bridgeerrorhandler":
case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
- case "cluster": target.setCluster(property(camelContext, com.datastax.driver.core.Cluster.class, value)); return true;
case "clustername":
case "clusterName": target.setClusterName(property(camelContext, java.lang.String.class, value)); return true;
case "consistencylevel":
- case "consistencyLevel": target.setConsistencyLevel(property(camelContext, com.datastax.driver.core.ConsistencyLevel.class, value)); return true;
+ case "consistencyLevel": target.setConsistencyLevel(property(camelContext, com.datastax.oss.driver.api.core.DefaultConsistencyLevel.class, value)); return true;
case "cql": target.setCql(property(camelContext, java.lang.String.class, value)); return true;
+ case "datacenter": target.setDatacenter(property(camelContext, java.lang.String.class, value)); return true;
case "delay": target.setDelay(property(camelContext, long.class, value)); return true;
case "exceptionhandler":
case "exceptionHandler": target.setExceptionHandler(property(camelContext, org.apache.camel.spi.ExceptionHandler.class, value)); return true;
@@ -45,8 +45,8 @@ public class CassandraEndpointConfigurer extends PropertyConfigurerSupport imple
case "initialDelay": target.setInitialDelay(property(camelContext, long.class, value)); return true;
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
- case "loadbalancingpolicy":
- case "loadBalancingPolicy": target.setLoadBalancingPolicy(property(camelContext, java.lang.String.class, value)); return true;
+ case "loadbalancingpolicyclass":
+ case "loadBalancingPolicyClass": target.setLoadBalancingPolicyClass(property(camelContext, java.lang.String.class, value)); return true;
case "password": target.setPassword(property(camelContext, java.lang.String.class, value)); return true;
case "pollstrategy":
case "pollStrategy": target.setPollStrategy(property(camelContext, org.apache.camel.spi.PollingConsumerPollStrategy.class, value)); return true;
@@ -65,7 +65,7 @@ public class CassandraEndpointConfigurer extends PropertyConfigurerSupport imple
case "schedulerProperties": target.setSchedulerProperties(property(camelContext, java.util.Map.class, value)); return true;
case "sendemptymessagewhenidle":
case "sendEmptyMessageWhenIdle": target.setSendEmptyMessageWhenIdle(property(camelContext, boolean.class, value)); return true;
- case "session": target.setSession(property(camelContext, com.datastax.driver.core.Session.class, value)); return true;
+ case "session": target.setSession(property(camelContext, com.datastax.oss.driver.api.core.CqlSession.class, value)); return true;
case "startscheduler":
case "startScheduler": target.setStartScheduler(property(camelContext, boolean.class, value)); return true;
case "synchronous": target.setSynchronous(property(camelContext, boolean.class, value)); return true;
@@ -86,17 +86,17 @@ public class CassandraEndpointConfigurer extends PropertyConfigurerSupport imple
answer.put("backoffMultiplier", int.class);
answer.put("basicPropertyBinding", boolean.class);
answer.put("bridgeErrorHandler", boolean.class);
- answer.put("cluster", com.datastax.driver.core.Cluster.class);
answer.put("clusterName", java.lang.String.class);
- answer.put("consistencyLevel", com.datastax.driver.core.ConsistencyLevel.class);
+ answer.put("consistencyLevel", com.datastax.oss.driver.api.core.DefaultConsistencyLevel.class);
answer.put("cql", java.lang.String.class);
+ answer.put("datacenter", java.lang.String.class);
answer.put("delay", long.class);
answer.put("exceptionHandler", org.apache.camel.spi.ExceptionHandler.class);
answer.put("exchangePattern", org.apache.camel.ExchangePattern.class);
answer.put("greedy", boolean.class);
answer.put("initialDelay", long.class);
answer.put("lazyStartProducer", boolean.class);
- answer.put("loadBalancingPolicy", java.lang.String.class);
+ answer.put("loadBalancingPolicyClass", java.lang.String.class);
answer.put("password", java.lang.String.class);
answer.put("pollStrategy", org.apache.camel.spi.PollingConsumerPollStrategy.class);
answer.put("prepareStatements", boolean.class);
@@ -107,7 +107,7 @@ public class CassandraEndpointConfigurer extends PropertyConfigurerSupport imple
answer.put("scheduler", java.lang.String.class);
answer.put("schedulerProperties", java.util.Map.class);
answer.put("sendEmptyMessageWhenIdle", boolean.class);
- answer.put("session", com.datastax.driver.core.Session.class);
+ answer.put("session", com.datastax.oss.driver.api.core.CqlSession.class);
answer.put("startScheduler", boolean.class);
answer.put("synchronous", boolean.class);
answer.put("timeUnit", java.util.concurrent.TimeUnit.class);
@@ -130,12 +130,12 @@ public class CassandraEndpointConfigurer extends PropertyConfigurerSupport imple
case "basicPropertyBinding": return target.isBasicPropertyBinding();
case "bridgeerrorhandler":
case "bridgeErrorHandler": return target.isBridgeErrorHandler();
- case "cluster": return target.getCluster();
case "clustername":
case "clusterName": return target.getClusterName();
case "consistencylevel":
case "consistencyLevel": return target.getConsistencyLevel();
case "cql": return target.getCql();
+ case "datacenter": return target.getDatacenter();
case "delay": return target.getDelay();
case "exceptionhandler":
case "exceptionHandler": return target.getExceptionHandler();
@@ -146,8 +146,8 @@ public class CassandraEndpointConfigurer extends PropertyConfigurerSupport imple
case "initialDelay": return target.getInitialDelay();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
- case "loadbalancingpolicy":
- case "loadBalancingPolicy": return target.getLoadBalancingPolicy();
+ case "loadbalancingpolicyclass":
+ case "loadBalancingPolicyClass": return target.getLoadBalancingPolicyClass();
case "password": return target.getPassword();
case "pollstrategy":
case "pollStrategy": return target.getPollStrategy();
diff --git a/components/camel-cassandraql/src/generated/resources/org/apache/camel/component/cassandra/cql.json b/components/camel-cassandraql/src/generated/resources/org/apache/camel/component/cassandra/cql.json
index 78315b5..44c9be1 100644
--- a/components/camel-cassandraql/src/generated/resources/org/apache/camel/component/cassandra/cql.json
+++ b/components/camel-cassandraql/src/generated/resources/org/apache/camel/component/cassandra/cql.json
@@ -30,15 +30,15 @@
"hosts": { "kind": "path", "displayName": "Hosts", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Hostname(s) cassansdra server(s). Multiple hosts can be separated by comma." },
"port": { "kind": "path", "displayName": "Port", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "description": "Port number of cassansdra server(s)" },
"keyspace": { "kind": "path", "displayName": "Keyspace", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Keyspace to use" },
- "cluster": { "kind": "parameter", "displayName": "Cluster", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.driver.core.Cluster", "deprecated": false, "secret": false, "description": "To use the Cluster instance (you would normally not use this option)" },
"clusterName": { "kind": "parameter", "displayName": "Cluster Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Cluster name" },
- "consistencyLevel": { "kind": "parameter", "displayName": "Consistency Level", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.driver.core.ConsistencyLevel", "enum": [ "ANY", "ONE", "TWO", "THREE", "QUORUM", "ALL", "LOCAL_QUORUM", "EACH_QUORUM", "SERIAL", "LOCAL_SERIAL", "LOCAL_ONE" ], "deprecated": false, "secret": false, "description": "Consistency level to use" },
+ "consistencyLevel": { "kind": "parameter", "displayName": "Consistency Level", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.oss.driver.api.core.DefaultConsistencyLevel", "enum": [ "ANY", "ONE", "TWO", "THREE", "QUORUM", "ALL", "LOCAL_ONE", "LOCAL_QUORUM", "EACH_QUORUM", "SERIAL", "LOCAL_SERIAL" ], "deprecated": false, "secret": false, "description": "Consistency level to use" },
"cql": { "kind": "parameter", "displayName": "Cql", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "CQL query to perform. Can be overridden with the message header with key CamelCqlQuery." },
- "loadBalancingPolicy": { "kind": "parameter", "displayName": "Load Balancing Policy", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "To use a specific LoadBalancingPolicy" },
+ "datacenter": { "kind": "parameter", "displayName": "Datacenter", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "datacenter1", "description": "Datacenter to use" },
+ "loadBalancingPolicyClass": { "kind": "parameter", "displayName": "Load Balancing Policy Class", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "To use a specific LoadBalancingPolicyClass" },
"password": { "kind": "parameter", "displayName": "Password", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Password for session authentication" },
"prepareStatements": { "kind": "parameter", "displayName": "Prepare Statements", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "description": "Whether to use PreparedStatements or regular Statements" },
"resultSetConversionStrategy": { "kind": "parameter", "displayName": "Result Set Conversion Strategy", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.cassandra.ResultSetConversionStrategy", "deprecated": false, "secret": false, "description": "To use a custom class that implements logic for converting ResultSet into message body ALL, ONE, LIMIT_10, LIMIT_100..." },
- "session": { "kind": "parameter", "displayName": "Session", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.driver.core.Session", "deprecated": false, "secret": false, "description": "To use the Session instance (you would normally not use this option)" },
+ "session": { "kind": "parameter", "displayName": "Session", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.datastax.oss.driver.api.core.CqlSession", "deprecated": false, "secret": false, "description": "To use the Session instance (you would normally not use this option)" },
"username": { "kind": "parameter", "displayName": "Username", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Username for session authentication" },
"bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled b [...]
"sendEmptyMessageWhenIdle": { "kind": "parameter", "displayName": "Send Empty Message When Idle", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead." },
diff --git a/components/camel-cassandraql/src/main/docs/cql-component.adoc b/components/camel-cassandraql/src/main/docs/cql-component.adoc
index 5c58bbc..c8f242a 100644
--- a/components/camel-cassandraql/src/main/docs/cql-component.adoc
+++ b/components/camel-cassandraql/src/main/docs/cql-component.adoc
@@ -106,15 +106,15 @@ with the following path and query parameters:
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
-| *cluster* (common) | To use the Cluster instance (you would normally not use this option) | | Cluster
| *clusterName* (common) | Cluster name | | String
-| *consistencyLevel* (common) | Consistency level to use. The value can be one of: ANY, ONE, TWO, THREE, QUORUM, ALL, LOCAL_QUORUM, EACH_QUORUM, SERIAL, LOCAL_SERIAL, LOCAL_ONE | | ConsistencyLevel
+| *consistencyLevel* (common) | Consistency level to use. The value can be one of: ANY, ONE, TWO, THREE, QUORUM, ALL, LOCAL_ONE, LOCAL_QUORUM, EACH_QUORUM, SERIAL, LOCAL_SERIAL | | DefaultConsistencyLevel
| *cql* (common) | CQL query to perform. Can be overridden with the message header with key CamelCqlQuery. | | String
-| *loadBalancingPolicy* (common) | To use a specific LoadBalancingPolicy | | String
+| *datacenter* (common) | Datacenter to use | datacenter1 | String
+| *loadBalancingPolicyClass* (common) | To use a specific LoadBalancingPolicyClass | | String
| *password* (common) | Password for session authentication | | String
| *prepareStatements* (common) | Whether to use PreparedStatements or regular Statements | true | boolean
| *resultSetConversionStrategy* (common) | To use a custom class that implements logic for converting ResultSet into message body ALL, ONE, LIMIT_10, LIMIT_100... | | ResultSetConversionStrategy
-| *session* (common) | To use the Session instance (you would normally not use this option) | | Session
+| *session* (common) | To use the Session instance (you would normally not use this option) | | CqlSession
| *username* (common) | Username for session authentication | | String
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *sendEmptyMessageWhenIdle* (consumer) | If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead. | false | boolean
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConsumer.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConsumer.java
index 20352ea..8beb685 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConsumer.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConsumer.java
@@ -16,9 +16,9 @@
*/
package org.apache.camel.component.cassandra;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Session;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
@@ -46,7 +46,7 @@ public class CassandraConsumer extends ScheduledPollConsumer {
@Override
protected int poll() throws Exception {
// Execute CQL Query
- Session session = getEndpoint().getSessionHolder().getSession();
+ CqlSession session = getEndpoint().getSessionHolder().getSession();
ResultSet resultSet;
if (isPrepareStatements()) {
resultSet = session.execute(preparedStatement.bind());
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraEndpoint.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraEndpoint.java
index f14e045..c5375e7 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraEndpoint.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraEndpoint.java
@@ -16,11 +16,17 @@
*/
package org.apache.camel.component.cassandra;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ConsistencyLevel;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Session;
+import java.net.InetSocketAddress;
+
+import com.datastax.oss.driver.api.core.ConsistencyLevel;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.CqlSessionBuilder;
+import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
+import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
+import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import org.apache.camel.Category;
import org.apache.camel.Component;
import org.apache.camel.Consumer;
@@ -32,9 +38,10 @@ import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.ScheduledPollEndpoint;
-import org.apache.camel.utils.cassandra.CassandraLoadBalancingPolicies;
import org.apache.camel.utils.cassandra.CassandraSessionHolder;
+
+
/**
* Integrate with Cassandra 2.0+ using the CQL3 API (not the Thrift API).
* Based on Cassandra Java Driver provided by DataStax.
@@ -52,6 +59,8 @@ public class CassandraEndpoint extends ScheduledPollEndpoint {
private Integer port;
@UriPath
private String keyspace;
+ @UriParam(defaultValue = "datacenter1")
+ private String datacenter = "datacenter1";
@UriParam
private String cql;
@UriParam(defaultValue = "true")
@@ -63,13 +72,11 @@ public class CassandraEndpoint extends ScheduledPollEndpoint {
@UriParam
private String password;
@UriParam
- private Cluster cluster;
- @UriParam
- private Session session;
+ private CqlSession session;
@UriParam
- private ConsistencyLevel consistencyLevel;
+ private DefaultConsistencyLevel consistencyLevel;
@UriParam
- private String loadBalancingPolicy;
+ private String loadBalancingPolicyClass;
@UriParam
private ResultSetConversionStrategy resultSetConversionStrategy = ResultSetConversionStrategies.all();
@@ -77,9 +84,8 @@ public class CassandraEndpoint extends ScheduledPollEndpoint {
super(endpointUri, component);
}
- public CassandraEndpoint(String uri, CassandraComponent component, Cluster cluster, Session session, String keyspace) {
+ public CassandraEndpoint(String uri, CassandraComponent component, CqlSession session, String keyspace) {
super(uri, component);
- this.cluster = cluster;
this.session = session;
this.keyspace = keyspace;
}
@@ -101,31 +107,22 @@ public class CassandraEndpoint extends ScheduledPollEndpoint {
super.doStart();
// we can get the cluster using various ways
-
- if (cluster == null && beanRef != null) {
+ if (session == null && beanRef != null) {
Object bean = CamelContextHelper.mandatoryLookup(getCamelContext(), beanRef);
- if (bean instanceof Session) {
- session = (Session)bean;
- cluster = session.getCluster();
- keyspace = session.getLoggedKeyspace();
- } else if (bean instanceof Cluster) {
- cluster = (Cluster)bean;
- session = null;
+ if (bean instanceof CqlSession) {
+ session = (CqlSession)bean;
+ keyspace = session.getKeyspace().isPresent() ? session.getKeyspace().get().toString() : null;
} else {
- throw new IllegalArgumentException("CQL Bean type should be of type Session or Cluster but was " + bean);
+ throw new IllegalArgumentException("CQL Bean type should be of type CqlSession but was " + bean);
}
}
- if (cluster == null && hosts != null) {
- // use the cluster builder to create the cluster
- cluster = createClusterBuilder().build();
+ if (session == null && hosts != null) {
+ // use the session builder to create the cluster
+ session = createSessionBuilder().build();
}
- if (cluster != null) {
- sessionHolder = new CassandraSessionHolder(cluster, keyspace);
- } else {
- sessionHolder = new CassandraSessionHolder(session);
- }
+ sessionHolder = new CassandraSessionHolder(session);
sessionHolder.start();
}
@@ -140,36 +137,33 @@ public class CassandraEndpoint extends ScheduledPollEndpoint {
return sessionHolder;
}
- protected Cluster.Builder createClusterBuilder() throws Exception {
- CassandraLoadBalancingPolicies cassLoadBalancingPolicies = new CassandraLoadBalancingPolicies();
- Cluster.Builder clusterBuilder = Cluster.builder();
+ protected CqlSessionBuilder createSessionBuilder() throws Exception {
+ CqlSessionBuilder sessionBuilder = CqlSession.builder();
for (String host : hosts.split(",")) {
- clusterBuilder = clusterBuilder.addContactPoint(host);
- }
- if (port != null) {
- clusterBuilder = clusterBuilder.withPort(port);
- }
- if (clusterName != null) {
- clusterBuilder = clusterBuilder.withClusterName(clusterName);
+ sessionBuilder.addContactPoint(new InetSocketAddress(host, port == null ? 9042 : port));
}
if (username != null && !username.isEmpty() && password != null) {
- clusterBuilder.withCredentials(username, password);
+ sessionBuilder.withAuthCredentials(username, password);
}
- if (loadBalancingPolicy != null && !loadBalancingPolicy.isEmpty()) {
- clusterBuilder.withLoadBalancingPolicy(cassLoadBalancingPolicies.getLoadBalancingPolicy(loadBalancingPolicy));
+ if (loadBalancingPolicyClass != null && !loadBalancingPolicyClass.isEmpty()) {
+ DriverConfigLoader driverConfigLoader = DriverConfigLoader.programmaticBuilder()
+ .withString(DefaultDriverOption.LOAD_BALANCING_POLICY_CLASS, loadBalancingPolicyClass)
+ .build();
+ sessionBuilder.withConfigLoader(driverConfigLoader);
}
- return clusterBuilder;
+
+ sessionBuilder.withLocalDatacenter(datacenter);
+ sessionBuilder.withKeyspace(keyspace);
+ return sessionBuilder;
}
/**
* Create and configure a Prepared CQL statement
*/
protected PreparedStatement prepareStatement(String cql) {
- PreparedStatement preparedStatement = getSessionHolder().getSession().prepare(cql);
- if (consistencyLevel != null) {
- preparedStatement.setConsistencyLevel(consistencyLevel);
- }
- return preparedStatement;
+ SimpleStatement statement = SimpleStatement.builder(cql)
+ .setConsistencyLevel(consistencyLevel).build();
+ return getSessionHolder().getSession().prepare(statement);
}
/**
@@ -242,30 +236,30 @@ public class CassandraEndpoint extends ScheduledPollEndpoint {
this.keyspace = keyspace;
}
- public String getCql() {
- return cql;
+ public String getDatacenter() {
+ return datacenter;
}
/**
- * CQL query to perform. Can be overridden with the message header with key
- * CamelCqlQuery.
+ * Datacenter to use
*/
- public void setCql(String cql) {
- this.cql = cql;
+ public void setDatacenter(String datacenter) {
+ this.datacenter = datacenter;
}
- public Cluster getCluster() {
- return cluster;
+ public String getCql() {
+ return cql;
}
/**
- * To use the Cluster instance (you would normally not use this option)
+ * CQL query to perform. Can be overridden with the message header with key
+ * CamelCqlQuery.
*/
- public void setCluster(Cluster cluster) {
- this.cluster = cluster;
+ public void setCql(String cql) {
+ this.cql = cql;
}
- public Session getSession() {
+ public CqlSession getSession() {
if (session == null) {
return sessionHolder.getSession();
} else {
@@ -276,7 +270,7 @@ public class CassandraEndpoint extends ScheduledPollEndpoint {
/**
* To use the Session instance (you would normally not use this option)
*/
- public void setSession(Session session) {
+ public void setSession(CqlSession session) {
this.session = session;
}
@@ -320,7 +314,7 @@ public class CassandraEndpoint extends ScheduledPollEndpoint {
/**
* Consistency level to use
*/
- public void setConsistencyLevel(ConsistencyLevel consistencyLevel) {
+ public void setConsistencyLevel(DefaultConsistencyLevel consistencyLevel) {
this.consistencyLevel = consistencyLevel;
}
@@ -348,14 +342,15 @@ public class CassandraEndpoint extends ScheduledPollEndpoint {
}
/**
- * To use a specific LoadBalancingPolicy
+ * To use a specific LoadBalancingPolicyClass
*/
- public String getLoadBalancingPolicy() {
- return loadBalancingPolicy;
+ public String getLoadBalancingPolicyClass() {
+ return loadBalancingPolicyClass;
}
- public void setLoadBalancingPolicy(String loadBalancingPolicy) {
- this.loadBalancingPolicy = loadBalancingPolicy;
+ public void setLoadBalancingPolicyClass(String loadBalancingPolicyClass) {
+ this.loadBalancingPolicyClass = loadBalancingPolicyClass;
}
+
}
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraProducer.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraProducer.java
index b55fee0..648eb4d 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraProducer.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraProducer.java
@@ -18,10 +18,11 @@ package org.apache.camel.component.cassandra;
import java.util.Collection;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.RegularStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Session;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.querybuilder.condition.ConditionalStatement;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.support.DefaultProducer;
@@ -99,7 +100,7 @@ public class CassandraProducer extends DefaultProducer {
Object[] cqlParams = getCqlParams(message);
ResultSet resultSet;
- Session session = getEndpoint().getSessionHolder().getSession();
+ CqlSession session = getEndpoint().getSessionHolder().getSession();
if (isPrepareStatements()) {
resultSet = executePreparedStatement(session, messageCql, cqlParams);
} else {
@@ -111,7 +112,7 @@ public class CassandraProducer extends DefaultProducer {
/**
* Execute CQL as PreparedStatement
*/
- private ResultSet executePreparedStatement(Session session, Object messageCql, Object[] cqlParams) {
+ private ResultSet executePreparedStatement(CqlSession session, Object messageCql, Object[] cqlParams) {
ResultSet resultSet;
PreparedStatement lPreparedStatement;
if (messageCql == null) {
@@ -120,9 +121,9 @@ public class CassandraProducer extends DefaultProducer {
} else if (messageCql instanceof String) {
// Message CQL
lPreparedStatement = getEndpoint().prepareStatement((String)messageCql);
- } else if (messageCql instanceof RegularStatement) {
+ } else if (messageCql instanceof SimpleStatement) {
// Message Statement
- lPreparedStatement = getEndpoint().getSession().prepare((RegularStatement)messageCql);
+ lPreparedStatement = getEndpoint().getSession().prepare((SimpleStatement) messageCql);
} else {
throw new IllegalArgumentException("Invalid " + CassandraConstants.CQL_QUERY + " header");
}
@@ -137,19 +138,19 @@ public class CassandraProducer extends DefaultProducer {
/**
* Execute CQL as is
*/
- private ResultSet executeStatement(Session session, Object messageCql, Object[] cqlParams) {
- ResultSet resultSet;
+ private ResultSet executeStatement(CqlSession session, Object messageCql, Object[] cqlParams) {
+ ResultSet resultSet = null;
String cql = null;
- RegularStatement statement = null;
+ SimpleStatement statement = null;
if (messageCql == null) {
// URI CQL
cql = getEndpoint().getCql();
} else if (messageCql instanceof String) {
// Message CQL
cql = (String)messageCql;
- } else if (messageCql instanceof RegularStatement) {
+ } else if (messageCql instanceof SimpleStatement) {
// Message Statement
- statement = (RegularStatement)messageCql;
+ statement = (SimpleStatement) messageCql;
} else {
throw new IllegalArgumentException("Invalid " + CassandraConstants.CQL_QUERY + " header");
}
@@ -158,7 +159,8 @@ public class CassandraProducer extends DefaultProducer {
} else if (isEmpty(cqlParams)) {
resultSet = session.execute(cql);
} else {
- resultSet = session.execute(cql, cqlParams);
+ resultSet = session.execute(
+ SimpleStatement.builder(cql).addPositionalValues(cqlParams).build());
}
return resultSet;
}
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategies.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategies.java
index dc2b8cc..ea8223b 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategies.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategies.java
@@ -22,8 +22,8 @@ import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
/**
* Implementations of {@link ResultSetConversionStrategy}
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategy.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategy.java
index 6f85243..120f162 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategy.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategy.java
@@ -16,7 +16,8 @@
*/
package org.apache.camel.component.cassandra;
-import com.datastax.driver.core.ResultSet;
+
+import com.datastax.oss.driver.api.core.cql.ResultSet;
/**
* Strategy to convert {@link ResultSet} into message body
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepository.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepository.java
index 3b6f458..e387d14 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepository.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepository.java
@@ -23,14 +23,14 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ConsistencyLevel;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.querybuilder.Delete;
-import com.datastax.driver.core.querybuilder.Insert;
-import com.datastax.driver.core.querybuilder.Select;
+import com.datastax.oss.driver.api.core.ConsistencyLevel;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.querybuilder.delete.Delete;
+import com.datastax.oss.driver.api.querybuilder.insert.Insert;
+import com.datastax.oss.driver.api.querybuilder.select.Select;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.spi.AggregationRepository;
@@ -40,8 +40,7 @@ import org.apache.camel.utils.cassandra.CassandraSessionHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker;
import static org.apache.camel.utils.cassandra.CassandraUtils.append;
import static org.apache.camel.utils.cassandra.CassandraUtils.applyConsistencyLevel;
import static org.apache.camel.utils.cassandra.CassandraUtils.concat;
@@ -127,14 +126,10 @@ public class CassandraAggregationRepository extends ServiceSupport implements Re
public CassandraAggregationRepository() {
}
- public CassandraAggregationRepository(Session session) {
+ public CassandraAggregationRepository(CqlSession session) {
this.sessionHolder = new CassandraSessionHolder(session);
}
- public CassandraAggregationRepository(Cluster cluster, String keyspace) {
- this.sessionHolder = new CassandraSessionHolder(cluster, keyspace);
- }
-
/**
* Generate primary key values from aggregation key.
*/
@@ -175,9 +170,9 @@ public class CassandraAggregationRepository extends ServiceSupport implements Re
private void initInsertStatement() {
Insert insert = generateInsert(table, getAllColumns(), false, ttl);
- insert = applyConsistencyLevel(insert, writeConsistencyLevel);
- LOGGER.debug("Generated Insert {}", insert);
- insertStatement = getSession().prepare(insert);
+ SimpleStatement statement = applyConsistencyLevel(insert.build(), writeConsistencyLevel);
+ LOGGER.debug("Generated Insert {}", statement);
+ insertStatement = getSession().prepare(statement);
}
/**
@@ -202,9 +197,9 @@ public class CassandraAggregationRepository extends ServiceSupport implements Re
protected void initSelectStatement() {
Select select = generateSelect(table, getAllColumns(), pkColumns);
- select = applyConsistencyLevel(select, readConsistencyLevel);
- LOGGER.debug("Generated Select {}", select);
- selectStatement = getSession().prepare(select);
+ SimpleStatement statement = applyConsistencyLevel(select.build(), readConsistencyLevel);
+ LOGGER.debug("Generated Select {}", statement);
+ selectStatement = getSession().prepare(statement);
}
/**
@@ -218,7 +213,7 @@ public class CassandraAggregationRepository extends ServiceSupport implements Re
Exchange exchange = null;
if (row != null) {
try {
- exchange = exchangeCodec.unmarshallExchange(camelContext, row.getBytes(exchangeColumn));
+ exchange = exchangeCodec.unmarshallExchange(camelContext, row.getByteBuffer(exchangeColumn));
} catch (IOException iOException) {
throw new CassandraAggregationException("Failed to read exchange", exchange, iOException);
} catch (ClassNotFoundException classNotFoundException) {
@@ -232,10 +227,10 @@ public class CassandraAggregationRepository extends ServiceSupport implements Re
// Confirm exchange in repository
private void initDeleteIfIdStatement() {
Delete delete = generateDelete(table, pkColumns, false);
- Delete.Conditions deleteIf = delete.onlyIf(eq(exchangeIdColumn, bindMarker()));
- deleteIf = applyConsistencyLevel(deleteIf, writeConsistencyLevel);
- LOGGER.debug("Generated Delete If Id {}", deleteIf);
- deleteIfIdStatement = getSession().prepare(deleteIf);
+ Delete deleteIf = delete.ifColumn(exchangeIdColumn).isEqualTo(bindMarker());
+ SimpleStatement statement = applyConsistencyLevel(deleteIf.build(), writeConsistencyLevel);
+ LOGGER.debug("Generated Delete If Id {}", statement);
+ deleteIfIdStatement = getSession().prepare(statement);
}
/**
@@ -261,9 +256,9 @@ public class CassandraAggregationRepository extends ServiceSupport implements Re
private void initDeleteStatement() {
Delete delete = generateDelete(table, pkColumns, false);
- delete = applyConsistencyLevel(delete, writeConsistencyLevel);
- LOGGER.debug("Generated Delete {}", delete);
- deleteStatement = getSession().prepare(delete);
+ SimpleStatement statement = applyConsistencyLevel(delete.build(), writeConsistencyLevel);
+ LOGGER.debug("Generated Delete {}", statement);
+ deleteStatement = getSession().prepare(statement);
}
/**
@@ -287,9 +282,9 @@ public class CassandraAggregationRepository extends ServiceSupport implements Re
// fixed
// PK
// columns
- select = applyConsistencyLevel(select, readConsistencyLevel);
- LOGGER.debug("Generated Select keys {}", select);
- selectKeyIdStatement = getSession().prepare(select);
+ SimpleStatement statement = applyConsistencyLevel(select.build(), readConsistencyLevel);
+ LOGGER.debug("Generated Select keys {}", statement);
+ selectKeyIdStatement = getSession().prepare(statement);
}
protected List<Row> selectKeyIds() {
@@ -347,11 +342,11 @@ public class CassandraAggregationRepository extends ServiceSupport implements Re
// -------------------------------------------------------------------------
// Getters and Setters
- public Session getSession() {
+ public CqlSession getSession() {
return sessionHolder.getSession();
}
- public void setSession(Session session) {
+ public void setSession(CqlSession session) {
this.sessionHolder = new CassandraSessionHolder(session);
}
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepository.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepository.java
index 6f80e53..cba2c50 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepository.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepository.java
@@ -16,8 +16,8 @@
*/
package org.apache.camel.processor.aggregate.cassandra;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.Session;
+
+import com.datastax.oss.driver.api.core.CqlSession;
/**
* Concrete implementation of {@link CassandraAggregationRepository} using 2
@@ -29,18 +29,12 @@ public class NamedCassandraAggregationRepository extends CassandraAggregationRep
setName("DEFAULT");
}
- public NamedCassandraAggregationRepository(Session session, String name) {
+ NamedCassandraAggregationRepository(CqlSession session, String name) {
super(session);
setPKColumns("NAME", "KEY");
setName(name);
}
- public NamedCassandraAggregationRepository(Cluster cluster, String keyspace, String name) {
- super(cluster, keyspace);
- setPKColumns("NAME", "KEY");
- setName(name);
- }
-
public String getName() {
return (String)getPrefixPKValues()[0];
}
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepository.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepository.java
index 61f8ce8..3347e94 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepository.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepository.java
@@ -16,16 +16,16 @@
*/
package org.apache.camel.processor.idempotent.cassandra;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ConsistencyLevel;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.querybuilder.Delete;
-import com.datastax.driver.core.querybuilder.Insert;
-import com.datastax.driver.core.querybuilder.Select;
-import com.datastax.driver.core.querybuilder.Truncate;
+import com.datastax.oss.driver.api.core.ConsistencyLevel;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.querybuilder.delete.Delete;
+import com.datastax.oss.driver.api.querybuilder.insert.Insert;
+import com.datastax.oss.driver.api.querybuilder.select.Select;
+import com.datastax.oss.driver.api.querybuilder.truncate.Truncate;
import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.utils.cassandra.CassandraSessionHolder;
@@ -87,14 +87,10 @@ public class CassandraIdempotentRepository extends ServiceSupport implements Ide
public CassandraIdempotentRepository() {
}
- public CassandraIdempotentRepository(Session session) {
+ public CassandraIdempotentRepository(CqlSession session) {
this.sessionHolder = new CassandraSessionHolder(session);
}
- public CassandraIdempotentRepository(Cluster cluster, String keyspace) {
- this.sessionHolder = new CassandraSessionHolder(cluster, keyspace);
- }
-
private boolean isKey(ResultSet resultSet) {
Row row = resultSet.one();
if (row == null) {
@@ -108,7 +104,7 @@ public class CassandraIdempotentRepository extends ServiceSupport implements Ide
protected final boolean isApplied(ResultSet resultSet) {
Row row = resultSet.one();
- return row == null || row.getBool("[applied]");
+ return row == null || row.getBoolean("[applied]");
}
protected Object[] getPKValues(String key) {
@@ -135,9 +131,9 @@ public class CassandraIdempotentRepository extends ServiceSupport implements Ide
protected void initInsertStatement() {
Insert insert = generateInsert(table, pkColumns, true, ttl);
- insert = applyConsistencyLevel(insert, writeConsistencyLevel);
- LOGGER.debug("Generated Insert {}", insert);
- insertStatement = getSession().prepare(insert);
+ SimpleStatement statement = applyConsistencyLevel(insert.build(), writeConsistencyLevel);
+ LOGGER.debug("Generated Insert {}", statement);
+ insertStatement = getSession().prepare(statement);
}
@Override
@@ -152,9 +148,9 @@ public class CassandraIdempotentRepository extends ServiceSupport implements Ide
protected void initSelectStatement() {
Select select = generateSelect(table, pkColumns, pkColumns);
- select = applyConsistencyLevel(select, readConsistencyLevel);
- LOGGER.debug("Generated Select {}", select);
- selectStatement = getSession().prepare(select);
+ SimpleStatement statement = applyConsistencyLevel(select.build(), readConsistencyLevel);
+ LOGGER.debug("Generated Select {}", statement);
+ selectStatement = getSession().prepare(statement);
}
@Override
@@ -174,9 +170,9 @@ public class CassandraIdempotentRepository extends ServiceSupport implements Ide
protected void initDeleteStatement() {
Delete delete = generateDelete(table, pkColumns, true);
- delete = applyConsistencyLevel(delete, writeConsistencyLevel);
- LOGGER.debug("Generated Delete {}", delete);
- deleteStatement = getSession().prepare(delete);
+ SimpleStatement statement = applyConsistencyLevel(delete.build(), writeConsistencyLevel);
+ LOGGER.debug("Generated Delete {}", statement);
+ deleteStatement = getSession().prepare(statement);
}
@Override
@@ -186,14 +182,15 @@ public class CassandraIdempotentRepository extends ServiceSupport implements Ide
return isApplied(getSession().execute(deleteStatement.bind(idValues)));
}
+
// -------------------------------------------------------------------------
// Clear the repository
protected void initClearStatement() {
Truncate truncate = generateTruncate(table);
- truncate = applyConsistencyLevel(truncate, writeConsistencyLevel);
- LOGGER.debug("Generated truncate for clear operation {}", truncate);
- truncateStatement = getSession().prepare(truncate);
+ SimpleStatement statement = applyConsistencyLevel(truncate.build(), writeConsistencyLevel);
+ LOGGER.debug("Generated truncate for clear operation {}", statement);
+ truncateStatement = getSession().prepare(statement);
}
@Override
@@ -205,11 +202,11 @@ public class CassandraIdempotentRepository extends ServiceSupport implements Ide
// -------------------------------------------------------------------------
// Getters & Setters
- public Session getSession() {
+ public CqlSession getSession() {
return sessionHolder.getSession();
}
- public void setSession(Session session) {
+ public void setSession(CqlSession session) {
this.sessionHolder = new CassandraSessionHolder(session);
}
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepository.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepository.java
index c279ecd..e1d6856 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepository.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepository.java
@@ -16,8 +16,8 @@
*/
package org.apache.camel.processor.idempotent.cassandra;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.Session;
+
+import com.datastax.oss.driver.api.core.CqlSession;
/**
* Concrete implementation of {@link CassandraIdempotentRepository} using 2
@@ -29,18 +29,12 @@ public class NamedCassandraIdempotentRepository extends CassandraIdempotentRepos
setName("DEFAULT");
}
- public NamedCassandraIdempotentRepository(Session session, String name) {
+ public NamedCassandraIdempotentRepository(CqlSession session, String name) {
super(session);
setPKColumns("NAME", "KEY");
setName(name);
}
- public NamedCassandraIdempotentRepository(Cluster cluster, String keyspace, String name) {
- super(cluster, keyspace);
- setPKColumns("NAME", "KEY");
- setName(name);
- }
-
public String getName() {
return getPrefixPKValues()[0];
}
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraLoadBalancingPolicies.java b/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraLoadBalancingPolicies.java
deleted file mode 100644
index f5bbd9a..0000000
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraLoadBalancingPolicies.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.utils.cassandra;
-
-import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
-import com.datastax.driver.core.policies.ErrorAwarePolicy;
-import com.datastax.driver.core.policies.LatencyAwarePolicy;
-import com.datastax.driver.core.policies.LoadBalancingPolicy;
-import com.datastax.driver.core.policies.RoundRobinPolicy;
-import com.datastax.driver.core.policies.TokenAwarePolicy;
-
-public class CassandraLoadBalancingPolicies {
-
- public final String roundRobinPolicy = "RoundRobinPolicy";
- public final String tokenAwarePolicy = "TokenAwarePolicy";
- public final String dcAwareRoundRobinPolicy = "DcAwareRoundRobinPolicy";
- public final String latencyAwarePolicy = "LatencyAwarePolicy";
- public final String errorAwarePolicy = "ErrorAwarePolicy";
-
- public LoadBalancingPolicy getLoadBalancingPolicy(String policy) {
- LoadBalancingPolicy loadBalancingPolicy = new RoundRobinPolicy();
- switch (policy) {
- case roundRobinPolicy:
- loadBalancingPolicy = new RoundRobinPolicy();
- break;
- case tokenAwarePolicy:
- loadBalancingPolicy = new TokenAwarePolicy(new RoundRobinPolicy());
- break;
- case dcAwareRoundRobinPolicy:
- loadBalancingPolicy = DCAwareRoundRobinPolicy.builder().build();
- break;
- case latencyAwarePolicy:
- loadBalancingPolicy = LatencyAwarePolicy.builder(new RoundRobinPolicy()).build();
- break;
- case errorAwarePolicy:
- loadBalancingPolicy = ErrorAwarePolicy.builder(new RoundRobinPolicy()).build();
- break;
- default:
- throw new IllegalArgumentException("Cassandra load balancing policy can be " + roundRobinPolicy + " ," + tokenAwarePolicy + " ," + dcAwareRoundRobinPolicy);
- }
- return loadBalancingPolicy;
- }
-}
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraSessionHolder.java b/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraSessionHolder.java
index c556701..165af30 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraSessionHolder.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraSessionHolder.java
@@ -16,21 +16,16 @@
*/
package org.apache.camel.utils.cassandra;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.Session;
+import com.datastax.oss.driver.api.core.CqlSession;
/**
* Holds a Cassandra Session and manages its lifecycle
*/
public class CassandraSessionHolder {
/**
- * Cluster
- */
- private final Cluster cluster;
- /**
* Session
*/
- private Session session;
+ private CqlSession session;
/**
* Keyspace name
*/
@@ -40,27 +35,12 @@ public class CassandraSessionHolder {
*/
private final boolean managedSession;
- public CassandraSessionHolder(Cluster cluster, String keyspace) {
- this.cluster = cluster;
- this.keyspace = keyspace;
- this.managedSession = true;
- }
-
- public CassandraSessionHolder(Session session) {
- this.cluster = session.getCluster();
+ public CassandraSessionHolder(CqlSession session) {
this.session = session;
- this.keyspace = session.getLoggedKeyspace();
this.managedSession = false;
}
public void start() {
- if (managedSession) {
- if (keyspace == null) {
- this.session = cluster.connect();
- } else {
- this.session = cluster.connect(keyspace);
- }
- }
}
public void stop() {
@@ -70,16 +50,7 @@ public class CassandraSessionHolder {
}
}
- public Session getSession() {
+ public CqlSession getSession() {
return session;
}
-
- public Cluster getCluster() {
- return cluster;
- }
-
- public String getKeyspace() {
- return keyspace;
- }
-
}
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraUtils.java b/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraUtils.java
index b0166f9..af9efaf 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraUtils.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraUtils.java
@@ -15,21 +15,22 @@
* limitations under the License.
*/
package org.apache.camel.utils.cassandra;
-
-import com.datastax.driver.core.ConsistencyLevel;
-import com.datastax.driver.core.RegularStatement;
-import com.datastax.driver.core.querybuilder.Delete;
-import com.datastax.driver.core.querybuilder.Insert;
-import com.datastax.driver.core.querybuilder.QueryBuilder;
-import com.datastax.driver.core.querybuilder.Select;
-import com.datastax.driver.core.querybuilder.Truncate;
-
-import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.delete;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.ttl;
+//
+import com.datastax.oss.driver.api.core.ConsistencyLevel;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
+import com.datastax.oss.driver.api.querybuilder.delete.Delete;
+import com.datastax.oss.driver.api.querybuilder.delete.DeleteSelection;
+import com.datastax.oss.driver.api.querybuilder.insert.Insert;
+import com.datastax.oss.driver.api.querybuilder.insert.InsertInto;
+import com.datastax.oss.driver.api.querybuilder.insert.RegularInsert;
+import com.datastax.oss.driver.api.querybuilder.select.Select;
+import com.datastax.oss.driver.api.querybuilder.select.SelectFrom;
+import com.datastax.oss.driver.api.querybuilder.truncate.Truncate;
+
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.insertInto;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.selectFrom;
public final class CassandraUtils {
@@ -101,17 +102,19 @@ public final class CassandraUtils {
* Generate Insert CQL.
*/
public static Insert generateInsert(String table, String[] columns, boolean ifNotExists, Integer ttl) {
- Insert insert = insertInto(table);
+ InsertInto into = insertInto(table);
+ RegularInsert regularInsert = null;
+ Insert insert = null;
for (String column : columns) {
- insert = insert.value(column, bindMarker());
+ regularInsert = (regularInsert != null ? regularInsert : into).value(column, bindMarker());
}
if (ifNotExists) {
- insert = insert.ifNotExists();
+ insert = regularInsert.ifNotExists();
}
if (ttl != null) {
- insert.using(ttl(ttl));
+ insert = (insert != null ? insert : regularInsert).usingTtl(ttl);
}
- return insert;
+ return insert != null ? insert : regularInsert;
}
/**
@@ -125,34 +128,47 @@ public final class CassandraUtils {
* Generate select where columns = ? CQL.
*/
public static Select generateSelect(String table, String[] selectColumns, String[] whereColumns, int whereColumnsMaxIndex) {
- Select select = select(selectColumns).from(table);
+ SelectFrom from = selectFrom(table);
+ Select select = null;
+ for (String column: selectColumns) {
+ select = (select != null ? select : from).column(column);
+ }
+ if (select == null) {
+ select = from.all();
+ }
if (isWhereClause(whereColumns, whereColumnsMaxIndex)) {
- Select.Where where = select.where();
for (int i = 0; i < whereColumns.length && i < whereColumnsMaxIndex; i++) {
- where.and(eq(whereColumns[i], bindMarker()));
+ select = select.whereColumn(whereColumns[i]).isEqualTo(bindMarker());
}
}
return select;
}
- /**
+ /**
* Generate delete where columns = ? CQL.
*/
public static Delete generateDelete(String table, String[] whereColumns, boolean ifExists) {
return generateDelete(table, whereColumns, size(whereColumns), ifExists);
}
- /**
+ /**
* Generate delete where columns = ? CQL.
*/
public static Delete generateDelete(String table, String[] whereColumns, int whereColumnsMaxIndex, boolean ifExists) {
- Delete delete = delete().from(table);
+ DeleteSelection deleteSelection = QueryBuilder.deleteFrom(table);
+ Delete delete = null;
+
if (isWhereClause(whereColumns, whereColumnsMaxIndex)) {
- Delete.Where where = delete.where();
for (int i = 0; i < whereColumns.length && i < whereColumnsMaxIndex; i++) {
- where.and(eq(whereColumns[i], bindMarker()));
+ delete = (delete != null ? delete : deleteSelection).whereColumn(whereColumns[i]).isEqualTo(bindMarker());
}
+ } else {
+ // Once there is at least one relation, the statement can be built
+ //(see https://docs.datastax.com/en/developer/java-driver/4.6/manual/query_builder/delete/#relations)
+ throw new IllegalArgumentException("Invalid delete statement. There has to be at least one relation. "
+ + "To delete all records, use Truncate");
}
+
if (ifExists) {
delete = delete.ifExists();
}
@@ -174,7 +190,7 @@ public final class CassandraUtils {
/**
* Apply consistency level if provided, else leave default.
*/
- public static <T extends RegularStatement> T applyConsistencyLevel(T statement, ConsistencyLevel consistencyLevel) {
+ public static <T extends SimpleStatement> T applyConsistencyLevel(T statement, ConsistencyLevel consistencyLevel) {
if (consistencyLevel != null) {
statement.setConsistencyLevel(consistencyLevel);
}
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/BaseCassandraTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/BaseCassandraTest.java
index f5c9711..a0880e3 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/BaseCassandraTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/BaseCassandraTest.java
@@ -35,5 +35,4 @@ public abstract class BaseCassandraTest extends CamelTestSupport {
// ignore shutdown errors
}
}
-
}
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraCQLUnit.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraCQLUnit.java
index 44ab9f1..30558b9 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraCQLUnit.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraCQLUnit.java
@@ -16,21 +16,23 @@
*/
package org.apache.camel.component.cassandra;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.Session;
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.internal.core.session.DefaultSession;
import org.cassandraunit.CQLDataLoader;
import org.cassandraunit.dataset.CQLDataSet;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.BeforeAllCallback;
import org.junit.jupiter.api.extension.BeforeEachCallback;
import org.junit.jupiter.api.extension.ExtensionContext;
import static org.junit.jupiter.api.Assumptions.assumeTrue;
-public class CassandraCQLUnit implements BeforeAllCallback, BeforeEachCallback {
+public class CassandraCQLUnit implements BeforeAllCallback, BeforeEachCallback, AfterEachCallback {
- public Session session;
- public Cluster cluster;
+ public CqlSession session;
protected CQLDataSet dataSet;
protected String configurationFileName;
protected long startupTimeoutMillis = EmbeddedCassandraServerHelper.DEFAULT_STARTUP_TIMEOUT;
@@ -51,16 +53,22 @@ public class CassandraCQLUnit implements BeforeAllCallback, BeforeEachCallback {
} else {
EmbeddedCassandraServerHelper.startEmbeddedCassandra(startupTimeoutMillis);
}
-
- /* create structure and load data */
- cluster = EmbeddedCassandraServerHelper.getCluster();
- session = EmbeddedCassandraServerHelper.getSession();
}
@Override
public void beforeEach(ExtensionContext context) throws Exception {
+ /* create structure and load data */
+ session = CqlSession.builder().build();
CQLDataLoader dataLoader = new CQLDataLoader(session);
dataLoader.load(dataSet);
}
+ @Override
+ public void afterEach(ExtensionContext extensionContext) throws Exception {
+ try {
+ session.close();
+ } catch (Throwable e) {
+ // ignore close errors
+ }
+ }
}
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentBeanRefTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentBeanRefTest.java
index aa6dfe1..5929318 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentBeanRefTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentBeanRefTest.java
@@ -16,8 +16,6 @@
*/
package org.apache.camel.component.cassandra;
-import org.apache.camel.Produce;
-import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.spi.Registry;
import org.apache.camel.support.SimpleRegistry;
@@ -30,18 +28,13 @@ public class CassandraComponentBeanRefTest extends BaseCassandraTest {
public static final String CQL = "insert into camel_user(login, first_name, last_name) values (?, ?, ?)";
public static final String SESSION_URI = "cql:bean:cassandraSession?cql=" + CQL;
- public static final String CLUSTER_URI = "cql:bean:cassandraCluster/camel_ks?cql=" + CQL;
@RegisterExtension
static CassandraCQLUnit cassandra = CassandraUnitUtils.cassandraCQLUnit();
- @Produce("direct:input")
- ProducerTemplate producerTemplate;
-
@Override
protected Registry createCamelRegistry() throws Exception {
SimpleRegistry registry = new SimpleRegistry();
- registry.bind("cassandraCluster", cassandra.cluster);
registry.bind("cassandraSession", cassandra.session);
return registry;
}
@@ -51,7 +44,6 @@ public class CassandraComponentBeanRefTest extends BaseCassandraTest {
return new RouteBuilder() {
public void configure() {
from("direct:inputSession").to(SESSION_URI);
- from("direct:inputCluster").to(CLUSTER_URI);
}
};
}
@@ -63,13 +55,4 @@ public class CassandraComponentBeanRefTest extends BaseCassandraTest {
assertEquals(CassandraUnitUtils.KEYSPACE, endpoint.getKeyspace());
assertEquals(CQL, endpoint.getCql());
}
-
- @Test
- public void testCluster() throws Exception {
- CassandraEndpoint endpoint = getMandatoryEndpoint(CLUSTER_URI, CassandraEndpoint.class);
-
- assertEquals(CassandraUnitUtils.KEYSPACE, endpoint.getKeyspace());
- assertEquals(CQL, endpoint.getCql());
- }
-
}
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentConsumerTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentConsumerTest.java
index 557afde..11bc1e8 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentConsumerTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentConsumerTest.java
@@ -19,7 +19,7 @@ package org.apache.camel.component.cassandra;
import java.util.List;
import java.util.concurrent.TimeUnit;
-import com.datastax.driver.core.Row;
+import com.datastax.oss.driver.api.core.cql.Row;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerTest.java
index 8c8693c..9dfade5 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerTest.java
@@ -20,25 +20,23 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ConsistencyLevel;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.querybuilder.Update;
+import com.datastax.oss.driver.api.core.ConsistencyLevel;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
+import com.datastax.oss.driver.api.querybuilder.update.Update;
import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.set;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.update;
import static org.apache.camel.test.junit5.TestSupport.assertIsInstanceOf;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class CassandraComponentProducerTest extends BaseCassandraTest {
@@ -71,7 +69,7 @@ public class CassandraComponentProducerTest extends BaseCassandraTest {
from("direct:input").to("cql://localhost/camel_ks?cql=" + CQL);
from("direct:inputNoParameter").to("cql://localhost/camel_ks?cql=" + NO_PARAMETER_CQL);
- from("direct:loadBalancingPolicy").to("cql://localhost/camel_ks?cql=" + NO_PARAMETER_CQL + "&loadBalancingPolicy=RoundRobinPolicy");
+ from("direct:loadBalancingPolicy").to("cql://localhost/camel_ks?cql=" + NO_PARAMETER_CQL + "&loadBalancingPolicyClass=org.apache.camel.component.cassandra.MockLoadBalancingPolicy");
from("direct:inputNotConsistent").to(NOT_CONSISTENT_URI);
from("direct:inputNoEndpointCql").to("cql://localhost/camel_ks");
}
@@ -82,15 +80,13 @@ public class CassandraComponentProducerTest extends BaseCassandraTest {
public void testRequestUriCql() throws Exception {
producerTemplate.requestBody(Arrays.asList("w_jiang", "Willem", "Jiang"));
- Cluster cluster = CassandraUnitUtils.cassandraCluster();
- Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
- ResultSet resultSet = session.execute("select login, first_name, last_name from camel_user where login = ?", "w_jiang");
+ CqlSession session = CassandraUnitUtils.cassandraSession();
+ ResultSet resultSet = session.execute(String.format("select login, first_name, last_name from camel_user where login = '%s'", "w_jiang"));
Row row = resultSet.one();
assertNotNull(row);
assertEquals("Willem", row.getString("first_name"));
assertEquals("Jiang", row.getString("last_name"));
session.close();
- cluster.close();
}
@Test
@@ -114,31 +110,31 @@ public class CassandraComponentProducerTest extends BaseCassandraTest {
producerTemplate.requestBodyAndHeader(new Object[] {"Claus 2", "Ibsen 2", "c_ibsen"}, CassandraConstants.CQL_QUERY,
"update camel_user set first_name=?, last_name=? where login=?");
- Cluster cluster = CassandraUnitUtils.cassandraCluster();
- Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
- ResultSet resultSet = session.execute("select login, first_name, last_name from camel_user where login = ?", "c_ibsen");
+ CqlSession session = CassandraUnitUtils.cassandraSession();
+ ResultSet resultSet = session.execute(String.format("select login, first_name, last_name from camel_user where login = '%s'", "c_ibsen"));
Row row = resultSet.one();
assertNotNull(row);
assertEquals("Claus 2", row.getString("first_name"));
assertEquals("Ibsen 2", row.getString("last_name"));
session.close();
- cluster.close();
}
@Test
public void testLoadBalancing() throws Exception {
- loadBalancingPolicyTemplate.requestBodyAndHeader(new Object[] {"Claus 2", "Ibsen 2", "c_ibsen"}, CassandraConstants.CQL_QUERY,
- "update camel_user set first_name=?, last_name=? where login=?");
+ loadBalancingPolicyTemplate.requestBodyAndHeader(new Object[]{"Claus 2", "Ibsen 2", "c_ibsen"}, CassandraConstants.CQL_QUERY,
+ "update camel_user set first_name=?, last_name=? where login=?");
- Cluster cluster = CassandraUnitUtils.cassandraCluster();
- Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
- ResultSet resultSet = session.execute("select login, first_name, last_name from camel_user where login = ?", "c_ibsen");
+
+ CqlSession session = CassandraUnitUtils.cassandraSession();
+ ResultSet resultSet = session.execute(String.format("select login, first_name, last_name from camel_user where login = '%s'", "c_ibsen"));
Row row = resultSet.one();
assertNotNull(row);
assertEquals("Claus 2", row.getString("first_name"));
assertEquals("Ibsen 2", row.getString("last_name"));
session.close();
- cluster.close();
+
+
+ assertTrue(MockLoadBalancingPolicy.used);
}
/**
@@ -146,18 +142,20 @@ public class CassandraComponentProducerTest extends BaseCassandraTest {
*/
@Test
public void testRequestMessageStatement() throws Exception {
- Update.Where update = update("camel_user").with(set("first_name", bindMarker())).and(set("last_name", bindMarker())).where(eq("login", bindMarker()));
- producerTemplate.requestBodyAndHeader(new Object[] {"Claus 2", "Ibsen 2", "c_ibsen"}, CassandraConstants.CQL_QUERY, update);
- Cluster cluster = CassandraUnitUtils.cassandraCluster();
- Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
- ResultSet resultSet = session.execute("select login, first_name, last_name from camel_user where login = ?", "c_ibsen");
+ Update update = QueryBuilder.update("camel_user")
+ .setColumn("first_name", bindMarker())
+ .setColumn("last_name", bindMarker())
+ .whereColumn("login").isEqualTo(bindMarker());
+ producerTemplate.requestBodyAndHeader(new Object[] {"Claus 2", "Ibsen 2", "c_ibsen"}, CassandraConstants.CQL_QUERY, update.build());
+
+ CqlSession session = CassandraUnitUtils.cassandraSession();
+ ResultSet resultSet = session.execute(String.format("select login, first_name, last_name from camel_user where login = '%s'", "c_ibsen"));
Row row = resultSet.one();
assertNotNull(row);
assertEquals("Claus 2", row.getString("first_name"));
assertEquals("Ibsen 2", row.getString("last_name"));
session.close();
- cluster.close();
}
/**
@@ -167,28 +165,30 @@ public class CassandraComponentProducerTest extends BaseCassandraTest {
*/
@Test
public void testEndpointNoCqlParameter() throws Exception {
- Update.Where updateFirstName = update("camel_user").with(set("first_name", bindMarker())).where(eq("login", bindMarker()));
- producerTemplateNoEndpointCql.sendBodyAndHeader(new Object[] {"Claus 2", "c_ibsen"}, CassandraConstants.CQL_QUERY, updateFirstName);
+ Update update = QueryBuilder.update("camel_user")
+ .setColumn("first_name", bindMarker())
+ .whereColumn("login").isEqualTo(bindMarker());
+ producerTemplateNoEndpointCql.sendBodyAndHeader(new Object[] {"Claus 2", "c_ibsen"}, CassandraConstants.CQL_QUERY, update.build());
- Cluster cluster = CassandraUnitUtils.cassandraCluster();
- Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
- ResultSet resultSet1 = session.execute("select login, first_name, last_name from camel_user where login = ?", "c_ibsen");
+ CqlSession session = CassandraUnitUtils.cassandraSession();
+ ResultSet resultSet1 = session.execute(String.format("select login, first_name, last_name from camel_user where login = '%s'", "c_ibsen"));
Row row1 = resultSet1.one();
assertNotNull(row1);
assertEquals("Claus 2", row1.getString("first_name"));
assertEquals("Ibsen", row1.getString("last_name"));
- Update.Where updateLastName = update("camel_user").with(set("last_name", bindMarker())).where(eq("login", bindMarker()));
- producerTemplateNoEndpointCql.sendBodyAndHeader(new Object[] {"Ibsen 2", "c_ibsen"}, CassandraConstants.CQL_QUERY, updateLastName);
+ update = QueryBuilder.update("camel_user")
+ .setColumn("last_name", bindMarker())
+ .whereColumn("login").isEqualTo(bindMarker());
+ producerTemplateNoEndpointCql.sendBodyAndHeader(new Object[] {"Ibsen 2", "c_ibsen"}, CassandraConstants.CQL_QUERY, update.build());
- ResultSet resultSet2 = session.execute("select login, first_name, last_name from camel_user where login = ?", "c_ibsen");
+ ResultSet resultSet2 = session.execute(String.format("select login, first_name, last_name from camel_user where login = '%s'", "c_ibsen"));
Row row2 = resultSet2.one();
assertNotNull(row2);
assertEquals("Claus 2", row2.getString("first_name"));
assertEquals("Ibsen 2", row2.getString("last_name"));
session.close();
- cluster.close();
}
@Test
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerUnpreparedTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerUnpreparedTest.java
index a52610a..c6a16fae 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerUnpreparedTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerUnpreparedTest.java
@@ -19,20 +19,19 @@ package org.apache.camel.component.cassandra;
import java.util.Arrays;
import java.util.List;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.querybuilder.Update;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
+import com.datastax.oss.driver.api.querybuilder.update.Update;
import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.set;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.update;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.literal;
import static org.apache.camel.test.junit5.TestSupport.assertIsInstanceOf;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -66,15 +65,13 @@ public class CassandraComponentProducerUnpreparedTest extends BaseCassandraTest
public void testRequestUriCql() throws Exception {
producerTemplate.requestBody(Arrays.asList("w_jiang", "Willem", "Jiang"));
- Cluster cluster = CassandraUnitUtils.cassandraCluster();
- Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
- ResultSet resultSet = session.execute("select login, first_name, last_name from camel_user where login = ?", "w_jiang");
+ CqlSession session = CassandraUnitUtils.cassandraSession();
+ ResultSet resultSet = session.execute(String.format("select login, first_name, last_name from camel_user where login = '%s'", "w_jiang"));
Row row = resultSet.one();
assertNotNull(row);
assertEquals("Willem", row.getString("first_name"));
assertEquals("Jiang", row.getString("last_name"));
session.close();
- cluster.close();
}
@Test
@@ -98,15 +95,13 @@ public class CassandraComponentProducerUnpreparedTest extends BaseCassandraTest
producerTemplate.requestBodyAndHeader(new Object[] {"Claus 2", "Ibsen 2", "c_ibsen"}, CassandraConstants.CQL_QUERY,
"update camel_user set first_name=?, last_name=? where login=?");
- Cluster cluster = CassandraUnitUtils.cassandraCluster();
- Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
- ResultSet resultSet = session.execute("select login, first_name, last_name from camel_user where login = ?", "c_ibsen");
+ CqlSession session = CassandraUnitUtils.cassandraSession();
+ ResultSet resultSet = session.execute(String.format("select login, first_name, last_name from camel_user where login = '%s'", "c_ibsen"));
Row row = resultSet.one();
assertNotNull(row);
assertEquals("Claus 2", row.getString("first_name"));
assertEquals("Ibsen 2", row.getString("last_name"));
session.close();
- cluster.close();
}
/**
@@ -114,18 +109,19 @@ public class CassandraComponentProducerUnpreparedTest extends BaseCassandraTest
*/
@Test
public void testRequestMessageStatement() throws Exception {
- Update.Where update = update("camel_user").with(set("first_name", "Claus 2")).and(set("last_name", "Ibsen 2")).where(eq("login", "c_ibsen"));
- producerTemplate.requestBodyAndHeader(null, CassandraConstants.CQL_QUERY, update);
-
- Cluster cluster = CassandraUnitUtils.cassandraCluster();
- Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
- ResultSet resultSet = session.execute("select login, first_name, last_name from camel_user where login = ?", "c_ibsen");
+ Update update = QueryBuilder.update("camel_user")
+ .setColumn("first_name", literal("Claus 2"))
+ .setColumn("last_name", literal("Ibsen 2"))
+ .whereColumn("login").isEqualTo(literal("c_ibsen"));
+ producerTemplate.requestBodyAndHeader(null, CassandraConstants.CQL_QUERY, update.build());
+
+ CqlSession session = CassandraUnitUtils.cassandraSession();
+ ResultSet resultSet = session.execute(String.format("select login, first_name, last_name from camel_user where login = '%s'", "c_ibsen"));
Row row = resultSet.one();
assertNotNull(row);
assertEquals("Claus 2", row.getString("first_name"));
assertEquals("Ibsen 2", row.getString("last_name"));
session.close();
- cluster.close();
}
}
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraUnitUtils.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraUnitUtils.java
index 3e11b40..fd85f54 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraUnitUtils.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraUnitUtils.java
@@ -16,7 +16,9 @@
*/
package org.apache.camel.component.cassandra;
-import com.datastax.driver.core.Cluster;
+import java.net.InetSocketAddress;
+
+import com.datastax.oss.driver.api.core.CqlSession;
import org.cassandraunit.dataset.CQLDataSet;
import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
@@ -68,7 +70,7 @@ public final class CassandraUnitUtils {
EmbeddedCassandraServerHelper.cleanEmbeddedCassandra();
}
- public static Cluster cassandraCluster() {
- return Cluster.builder().addContactPoint(HOST).withClusterName("camel-cluster").build();
+ public static CqlSession cassandraSession() {
+ return CqlSession.builder().addContactPoint(new InetSocketAddress(HOST, 9042)).withKeyspace(KEYSPACE).withLocalDatacenter("datacenter1").build();
}
}
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/MockLoadBalancingPolicy.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/MockLoadBalancingPolicy.java
new file mode 100644
index 0000000..00f62af
--- /dev/null
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/MockLoadBalancingPolicy.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.cassandra;
+
+import java.util.Queue;
+
+import com.datastax.oss.driver.api.core.context.DriverContext;
+import com.datastax.oss.driver.api.core.metadata.Node;
+import com.datastax.oss.driver.api.core.session.Request;
+import com.datastax.oss.driver.api.core.session.Session;
+import com.datastax.oss.driver.internal.core.loadbalancing.DefaultLoadBalancingPolicy;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
+
+public class MockLoadBalancingPolicy extends DefaultLoadBalancingPolicy {
+
+ public static boolean used;
+
+ public MockLoadBalancingPolicy(@NonNull DriverContext context, @NonNull String profileName) {
+ super(context, profileName);
+ }
+
+ @NonNull
+ @Override
+ public Queue<Node> newQueryPlan(@Nullable Request request, @Nullable Session session) {
+ MockLoadBalancingPolicy.used = true;
+ return super.newQueryPlan(request, session);
+ }
+}
+
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/ResultSetConversionStrategiesTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/ResultSetConversionStrategiesTest.java
index dcff03c..8856375 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/ResultSetConversionStrategiesTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/ResultSetConversionStrategiesTest.java
@@ -19,8 +19,8 @@ package org.apache.camel.component.cassandra;
import java.util.Collections;
import java.util.List;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
import org.junit.jupiter.api.Test;
import static org.junit.Assert.assertEquals;
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepositoryTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepositoryTest.java
index dc14dd7..09f7221 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepositoryTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepositoryTest.java
@@ -49,19 +49,19 @@ public class CassandraAggregationRepositoryTest extends BaseCassandraTest {
@Override
protected void doPreSetup() throws Exception {
camelContext = new DefaultCamelContext();
- aggregationRepository = new CassandraAggregationRepository(cassandra.cluster, CassandraUnitUtils.KEYSPACE);
+ aggregationRepository = new CassandraAggregationRepository(cassandra.session);
aggregationRepository.start();
super.doPreSetup();
}
- @Override
- @AfterEach
- public void tearDown() throws Exception {
- aggregationRepository.stop();
- }
+// @Override
+// @AfterEach
+// public void tearDown() throws Exception {
+// aggregationRepository.stop();
+// }
private boolean exists(String key) {
- return cassandra.session.execute("select KEY from CAMEL_AGGREGATION where KEY=?", key).one() != null;
+ return cassandra.session.execute(String.format("select KEY from CAMEL_AGGREGATION where KEY='%s'", key)).one() != null;
}
@Test
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationSerializedHeadersTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationSerializedHeadersTest.java
index 0129e81..f2f8697 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationSerializedHeadersTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationSerializedHeadersTest.java
@@ -40,7 +40,7 @@ public class CassandraAggregationSerializedHeadersTest extends BaseCassandraTest
@Override
protected void doPreSetup() throws Exception {
- aggregationRepository = new NamedCassandraAggregationRepository(cassandra.cluster, CassandraUnitUtils.KEYSPACE, "ID");
+ aggregationRepository = new NamedCassandraAggregationRepository(cassandra.session, "ID");
aggregationRepository.setTable("NAMED_CAMEL_AGGREGATION");
aggregationRepository.setAllowSerializedHeaders(true);
aggregationRepository.start();
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationTest.java
index 5f90153..059a7ef 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationTest.java
@@ -39,7 +39,7 @@ public class CassandraAggregationTest extends BaseCassandraTest {
@Override
protected void doPreSetup() throws Exception {
- aggregationRepository = new NamedCassandraAggregationRepository(cassandra.cluster, CassandraUnitUtils.KEYSPACE, "ID");
+ aggregationRepository = new NamedCassandraAggregationRepository(cassandra.session, "ID");
aggregationRepository.setTable("NAMED_CAMEL_AGGREGATION");
aggregationRepository.start();
super.doPreSetup();
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepositoryTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepositoryTest.java
index cf58242..17efe20 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepositoryTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepositoryTest.java
@@ -42,6 +42,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class NamedCassandraAggregationRepositoryTest extends BaseCassandraTest {
@RegisterExtension
+
static CassandraCQLUnit cassandra = CassandraUnitUtils.cassandraCQLUnit("NamedAggregationDataSet.cql");
private CassandraAggregationRepository aggregationRepository;
@@ -51,19 +52,19 @@ public class NamedCassandraAggregationRepositoryTest extends BaseCassandraTest {
@BeforeEach
public void setUp() throws Exception {
camelContext = new DefaultCamelContext();
- aggregationRepository = new NamedCassandraAggregationRepository(cassandra.cluster, CassandraUnitUtils.KEYSPACE, "ID");
+ aggregationRepository = new NamedCassandraAggregationRepository(cassandra.session, "ID");
aggregationRepository.setTable("NAMED_CAMEL_AGGREGATION");
aggregationRepository.start();
}
- @Override
- @AfterEach
- public void tearDown() throws Exception {
- aggregationRepository.stop();
- }
+// @Override
+// @AfterEach
+// public void tearDown() throws Exception {
+// aggregationRepository.stop();
+// }
private boolean exists(String key) {
- return cassandra.session.execute("select KEY from NAMED_CAMEL_AGGREGATION where NAME=? and KEY=?", "ID", key).one() != null;
+ return cassandra.session.execute(String.format("select KEY from NAMED_CAMEL_AGGREGATION where NAME='ID' and KEY='%s'", key)).one() != null;
}
@Test
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepositoryTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepositoryTest.java
index 5af2140c..6524386 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepositoryTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepositoryTest.java
@@ -40,7 +40,8 @@ public class CassandraIdempotentRepositoryTest extends BaseCassandraTest {
@Override
@BeforeEach
public void setUp() throws Exception {
- idempotentRepository = new CassandraIdempotentRepository(cassandra.cluster, CassandraUnitUtils.KEYSPACE);
+ idempotentRepository = new CassandraIdempotentRepository(cassandra.session);
+
idempotentRepository.start();
super.setUp();
}
@@ -53,7 +54,7 @@ public class CassandraIdempotentRepositoryTest extends BaseCassandraTest {
}
private boolean exists(String key) {
- return cassandra.session.execute("select KEY from CAMEL_IDEMPOTENT where KEY=?", key).one() != null;
+ return cassandra.session.execute(String.format("select KEY from CAMEL_IDEMPOTENT where KEY='%s'", key)).one() != null;
}
@Test
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentTest.java
index 53a0277..2559dee 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentTest.java
@@ -37,7 +37,7 @@ public class CassandraIdempotentTest extends BaseCassandraTest {
@Override
protected void doPreSetup() throws Exception {
- idempotentRepository = new NamedCassandraIdempotentRepository(cassandra.cluster, CassandraUnitUtils.KEYSPACE, "ID");
+ idempotentRepository = new NamedCassandraIdempotentRepository(cassandra.session, "ID");
idempotentRepository.setTable("NAMED_CAMEL_IDEMPOTENT");
idempotentRepository.start();
super.doPreSetup();
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepositoryTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepositoryTest.java
index a21ece8..09bc08c 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepositoryTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepositoryTest.java
@@ -40,7 +40,7 @@ public class NamedCassandraIdempotentRepositoryTest extends BaseCassandraTest {
@Override
@BeforeEach
public void setUp() throws Exception {
- idempotentRepository = new NamedCassandraIdempotentRepository(cassandra.cluster, CassandraUnitUtils.KEYSPACE, "ID");
+ idempotentRepository = new NamedCassandraIdempotentRepository(cassandra.session, "ID");
idempotentRepository.setTable("NAMED_CAMEL_IDEMPOTENT");
idempotentRepository.start();
super.setUp();
@@ -54,7 +54,7 @@ public class NamedCassandraIdempotentRepositoryTest extends BaseCassandraTest {
}
private boolean exists(String key) {
- return cassandra.session.execute("select KEY from NAMED_CAMEL_IDEMPOTENT where NAME=? and KEY=?", "ID", key).one() != null;
+ return cassandra.session.execute(String.format("select KEY from NAMED_CAMEL_IDEMPOTENT where NAME='ID' and KEY='%s'", key)).one() != null;
}
@Test
diff --git a/components/camel-cassandraql/src/test/resources/camel-cassandra.yaml b/components/camel-cassandraql/src/test/resources/camel-cassandra.yaml
index f6f856d..1c25727 100644
--- a/components/camel-cassandraql/src/test/resources/camel-cassandra.yaml
+++ b/components/camel-cassandraql/src/test/resources/camel-cassandra.yaml
@@ -129,6 +129,8 @@ data_file_directories:
# commitlog_directory: /var/lib/cassandra/commitlog
commitlog_directory: target/camel-cassandra/commitlog
+cdc_raw_directory: target/embeddedCassandra/cdc
+
# policy for data disk failures:
# stop_paranoid: shut down gossip and Thrift even for single-sstable errors.
# stop: shut down gossip and Thrift, leaving the node effectively dead, but
@@ -783,3 +785,4 @@ internode_compression: all
# reducing overhead from the TCP protocol itself, at the cost of increasing
# latency if you block for cross-datacenter responses.
inter_dc_tcp_nodelay: false
+
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java
index 4a11e12..fbb8d8b 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java
@@ -3634,7 +3634,7 @@ public class StaticEndpointBuilders {
*
* @param path beanRef:hosts:port/keyspace
*/
- public static org.apache.camel.builder.endpoint.dsl.CassandraEndpointBuilderFactory.CassandraEndpointBuilder cql(
+ static org.apache.camel.builder.endpoint.dsl.CassandraEndpointBuilderFactory.CassandraEndpointBuilder cql(
String path) {
return org.apache.camel.builder.endpoint.dsl.CassandraEndpointBuilderFactory.endpointBuilder("cql", path);
}
@@ -3665,7 +3665,7 @@ public class StaticEndpointBuilders {
* instead of the default name
* @param path beanRef:hosts:port/keyspace
*/
- public static org.apache.camel.builder.endpoint.dsl.CassandraEndpointBuilderFactory.CassandraEndpointBuilder cql(
+ static org.apache.camel.builder.endpoint.dsl.CassandraEndpointBuilderFactory.CassandraEndpointBuilder cql(
String componentName,
String path) {
return org.apache.camel.builder.endpoint.dsl.CassandraEndpointBuilderFactory.endpointBuilder(componentName, path);
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CassandraEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CassandraEndpointBuilderFactory.java
index 694705f..25dae08 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CassandraEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CassandraEndpointBuilderFactory.java
@@ -47,29 +47,6 @@ public interface CassandraEndpointBuilderFactory {
return (AdvancedCassandraEndpointConsumerBuilder) this;
}
/**
- * To use the Cluster instance (you would normally not use this option).
- *
- * The option is a: <code>com.datastax.driver.core.Cluster</code> type.
- *
- * Group: common
- */
- default CassandraEndpointConsumerBuilder cluster(Object cluster) {
- doSetProperty("cluster", cluster);
- return this;
- }
- /**
- * To use the Cluster instance (you would normally not use this option).
- *
- * The option will be converted to a
- * <code>com.datastax.driver.core.Cluster</code> type.
- *
- * Group: common
- */
- default CassandraEndpointConsumerBuilder cluster(String cluster) {
- doSetProperty("cluster", cluster);
- return this;
- }
- /**
* Cluster name.
*
* The option is a: <code>java.lang.String</code> type.
@@ -84,12 +61,13 @@ public interface CassandraEndpointBuilderFactory {
* Consistency level to use.
*
* The option is a:
- * <code>com.datastax.driver.core.ConsistencyLevel</code> type.
+ * <code>com.datastax.oss.driver.api.core.DefaultConsistencyLevel</code>
+ * type.
*
* Group: common
*/
default CassandraEndpointConsumerBuilder consistencyLevel(
- ConsistencyLevel consistencyLevel) {
+ DefaultConsistencyLevel consistencyLevel) {
doSetProperty("consistencyLevel", consistencyLevel);
return this;
}
@@ -97,7 +75,8 @@ public interface CassandraEndpointBuilderFactory {
* Consistency level to use.
*
* The option will be converted to a
- * <code>com.datastax.driver.core.ConsistencyLevel</code> type.
+ * <code>com.datastax.oss.driver.api.core.DefaultConsistencyLevel</code>
+ * type.
*
* Group: common
*/
@@ -119,15 +98,27 @@ public interface CassandraEndpointBuilderFactory {
return this;
}
/**
- * To use a specific LoadBalancingPolicy.
+ * Datacenter to use.
*
* The option is a: <code>java.lang.String</code> type.
*
+ * Default: datacenter1
* Group: common
*/
- default CassandraEndpointConsumerBuilder loadBalancingPolicy(
- String loadBalancingPolicy) {
- doSetProperty("loadBalancingPolicy", loadBalancingPolicy);
+ default CassandraEndpointConsumerBuilder datacenter(String datacenter) {
+ doSetProperty("datacenter", datacenter);
+ return this;
+ }
+ /**
+ * To use a specific LoadBalancingPolicyClass.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default CassandraEndpointConsumerBuilder loadBalancingPolicyClass(
+ String loadBalancingPolicyClass) {
+ doSetProperty("loadBalancingPolicyClass", loadBalancingPolicyClass);
return this;
}
/**
@@ -198,7 +189,8 @@ public interface CassandraEndpointBuilderFactory {
/**
* To use the Session instance (you would normally not use this option).
*
- * The option is a: <code>com.datastax.driver.core.Session</code> type.
+ * The option is a:
+ * <code>com.datastax.oss.driver.api.core.CqlSession</code> type.
*
* Group: common
*/
@@ -210,7 +202,7 @@ public interface CassandraEndpointBuilderFactory {
* To use the Session instance (you would normally not use this option).
*
* The option will be converted to a
- * <code>com.datastax.driver.core.Session</code> type.
+ * <code>com.datastax.oss.driver.api.core.CqlSession</code> type.
*
* Group: common
*/
@@ -837,29 +829,6 @@ public interface CassandraEndpointBuilderFactory {
return (AdvancedCassandraEndpointProducerBuilder) this;
}
/**
- * To use the Cluster instance (you would normally not use this option).
- *
- * The option is a: <code>com.datastax.driver.core.Cluster</code> type.
- *
- * Group: common
- */
- default CassandraEndpointProducerBuilder cluster(Object cluster) {
- doSetProperty("cluster", cluster);
- return this;
- }
- /**
- * To use the Cluster instance (you would normally not use this option).
- *
- * The option will be converted to a
- * <code>com.datastax.driver.core.Cluster</code> type.
- *
- * Group: common
- */
- default CassandraEndpointProducerBuilder cluster(String cluster) {
- doSetProperty("cluster", cluster);
- return this;
- }
- /**
* Cluster name.
*
* The option is a: <code>java.lang.String</code> type.
@@ -874,12 +843,13 @@ public interface CassandraEndpointBuilderFactory {
* Consistency level to use.
*
* The option is a:
- * <code>com.datastax.driver.core.ConsistencyLevel</code> type.
+ * <code>com.datastax.oss.driver.api.core.DefaultConsistencyLevel</code>
+ * type.
*
* Group: common
*/
default CassandraEndpointProducerBuilder consistencyLevel(
- ConsistencyLevel consistencyLevel) {
+ DefaultConsistencyLevel consistencyLevel) {
doSetProperty("consistencyLevel", consistencyLevel);
return this;
}
@@ -887,7 +857,8 @@ public interface CassandraEndpointBuilderFactory {
* Consistency level to use.
*
* The option will be converted to a
- * <code>com.datastax.driver.core.ConsistencyLevel</code> type.
+ * <code>com.datastax.oss.driver.api.core.DefaultConsistencyLevel</code>
+ * type.
*
* Group: common
*/
@@ -909,15 +880,27 @@ public interface CassandraEndpointBuilderFactory {
return this;
}
/**
- * To use a specific LoadBalancingPolicy.
+ * Datacenter to use.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: datacenter1
+ * Group: common
+ */
+ default CassandraEndpointProducerBuilder datacenter(String datacenter) {
+ doSetProperty("datacenter", datacenter);
+ return this;
+ }
+ /**
+ * To use a specific LoadBalancingPolicyClass.
*
* The option is a: <code>java.lang.String</code> type.
*
* Group: common
*/
- default CassandraEndpointProducerBuilder loadBalancingPolicy(
- String loadBalancingPolicy) {
- doSetProperty("loadBalancingPolicy", loadBalancingPolicy);
+ default CassandraEndpointProducerBuilder loadBalancingPolicyClass(
+ String loadBalancingPolicyClass) {
+ doSetProperty("loadBalancingPolicyClass", loadBalancingPolicyClass);
return this;
}
/**
@@ -988,7 +971,8 @@ public interface CassandraEndpointBuilderFactory {
/**
* To use the Session instance (you would normally not use this option).
*
- * The option is a: <code>com.datastax.driver.core.Session</code> type.
+ * The option is a:
+ * <code>com.datastax.oss.driver.api.core.CqlSession</code> type.
*
* Group: common
*/
@@ -1000,7 +984,7 @@ public interface CassandraEndpointBuilderFactory {
* To use the Session instance (you would normally not use this option).
*
* The option will be converted to a
- * <code>com.datastax.driver.core.Session</code> type.
+ * <code>com.datastax.oss.driver.api.core.CqlSession</code> type.
*
* Group: common
*/
@@ -1141,29 +1125,6 @@ public interface CassandraEndpointBuilderFactory {
return (AdvancedCassandraEndpointBuilder) this;
}
/**
- * To use the Cluster instance (you would normally not use this option).
- *
- * The option is a: <code>com.datastax.driver.core.Cluster</code> type.
- *
- * Group: common
- */
- default CassandraEndpointBuilder cluster(Object cluster) {
- doSetProperty("cluster", cluster);
- return this;
- }
- /**
- * To use the Cluster instance (you would normally not use this option).
- *
- * The option will be converted to a
- * <code>com.datastax.driver.core.Cluster</code> type.
- *
- * Group: common
- */
- default CassandraEndpointBuilder cluster(String cluster) {
- doSetProperty("cluster", cluster);
- return this;
- }
- /**
* Cluster name.
*
* The option is a: <code>java.lang.String</code> type.
@@ -1178,12 +1139,13 @@ public interface CassandraEndpointBuilderFactory {
* Consistency level to use.
*
* The option is a:
- * <code>com.datastax.driver.core.ConsistencyLevel</code> type.
+ * <code>com.datastax.oss.driver.api.core.DefaultConsistencyLevel</code>
+ * type.
*
* Group: common
*/
default CassandraEndpointBuilder consistencyLevel(
- ConsistencyLevel consistencyLevel) {
+ DefaultConsistencyLevel consistencyLevel) {
doSetProperty("consistencyLevel", consistencyLevel);
return this;
}
@@ -1191,7 +1153,8 @@ public interface CassandraEndpointBuilderFactory {
* Consistency level to use.
*
* The option will be converted to a
- * <code>com.datastax.driver.core.ConsistencyLevel</code> type.
+ * <code>com.datastax.oss.driver.api.core.DefaultConsistencyLevel</code>
+ * type.
*
* Group: common
*/
@@ -1213,15 +1176,27 @@ public interface CassandraEndpointBuilderFactory {
return this;
}
/**
- * To use a specific LoadBalancingPolicy.
+ * Datacenter to use.
*
* The option is a: <code>java.lang.String</code> type.
*
+ * Default: datacenter1
* Group: common
*/
- default CassandraEndpointBuilder loadBalancingPolicy(
- String loadBalancingPolicy) {
- doSetProperty("loadBalancingPolicy", loadBalancingPolicy);
+ default CassandraEndpointBuilder datacenter(String datacenter) {
+ doSetProperty("datacenter", datacenter);
+ return this;
+ }
+ /**
+ * To use a specific LoadBalancingPolicyClass.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default CassandraEndpointBuilder loadBalancingPolicyClass(
+ String loadBalancingPolicyClass) {
+ doSetProperty("loadBalancingPolicyClass", loadBalancingPolicyClass);
return this;
}
/**
@@ -1292,7 +1267,8 @@ public interface CassandraEndpointBuilderFactory {
/**
* To use the Session instance (you would normally not use this option).
*
- * The option is a: <code>com.datastax.driver.core.Session</code> type.
+ * The option is a:
+ * <code>com.datastax.oss.driver.api.core.CqlSession</code> type.
*
* Group: common
*/
@@ -1304,7 +1280,7 @@ public interface CassandraEndpointBuilderFactory {
* To use the Session instance (you would normally not use this option).
*
* The option will be converted to a
- * <code>com.datastax.driver.core.Session</code> type.
+ * <code>com.datastax.oss.driver.api.core.CqlSession</code> type.
*
* Group: common
*/
@@ -1392,21 +1368,22 @@ public interface CassandraEndpointBuilderFactory {
}
/**
- * Proxy enum for <code>com.datastax.driver.core.ConsistencyLevel</code>
+ * Proxy enum for
+ * <code>com.datastax.oss.driver.api.core.DefaultConsistencyLevel</code>
* enum.
*/
- enum ConsistencyLevel {
+ enum DefaultConsistencyLevel {
ANY,
ONE,
TWO,
THREE,
QUORUM,
ALL,
+ LOCAL_ONE,
LOCAL_QUORUM,
EACH_QUORUM,
SERIAL,
- LOCAL_SERIAL,
- LOCAL_ONE;
+ LOCAL_SERIAL;
}
public interface CassandraBuilders {
diff --git a/docs/components/modules/ROOT/pages/cql-component.adoc b/docs/components/modules/ROOT/pages/cql-component.adoc
index cc91c31..03f4507 100644
--- a/docs/components/modules/ROOT/pages/cql-component.adoc
+++ b/docs/components/modules/ROOT/pages/cql-component.adoc
@@ -108,15 +108,15 @@ with the following path and query parameters:
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
-| *cluster* (common) | To use the Cluster instance (you would normally not use this option) | | Cluster
| *clusterName* (common) | Cluster name | | String
-| *consistencyLevel* (common) | Consistency level to use. The value can be one of: ANY, ONE, TWO, THREE, QUORUM, ALL, LOCAL_QUORUM, EACH_QUORUM, SERIAL, LOCAL_SERIAL, LOCAL_ONE | | ConsistencyLevel
+| *consistencyLevel* (common) | Consistency level to use. The value can be one of: ANY, ONE, TWO, THREE, QUORUM, ALL, LOCAL_ONE, LOCAL_QUORUM, EACH_QUORUM, SERIAL, LOCAL_SERIAL | | DefaultConsistencyLevel
| *cql* (common) | CQL query to perform. Can be overridden with the message header with key CamelCqlQuery. | | String
-| *loadBalancingPolicy* (common) | To use a specific LoadBalancingPolicy | | String
+| *datacenter* (common) | Datacenter to use | datacenter1 | String
+| *loadBalancingPolicyClass* (common) | To use a specific LoadBalancingPolicyClass | | String
| *password* (common) | Password for session authentication | | String
| *prepareStatements* (common) | Whether to use PreparedStatements or regular Statements | true | boolean
| *resultSetConversionStrategy* (common) | To use a custom class that implements logic for converting ResultSet into message body ALL, ONE, LIMIT_10, LIMIT_100... | | ResultSetConversionStrategy
-| *session* (common) | To use the Session instance (you would normally not use this option) | | Session
+| *session* (common) | To use the Session instance (you would normally not use this option) | | CqlSession
| *username* (common) | Username for session authentication | | String
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *sendEmptyMessageWhenIdle* (consumer) | If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead. | false | boolean
diff --git a/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_5.adoc b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_5.adoc
index 950badc..a55d143 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_5.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_5.adoc
@@ -11,6 +11,20 @@ from both 3.0 to 3.1 and 3.1 to 3.2.
The `bean(class)` EIP will now lookup in the registry first whether there is a single bean instance of the given class type
and use the existing bean (for singleton scope) instead of creating a new instance.
+=== camel-cassandraql
+
+Load-balancing policy has changed. There are no existing load balancing policies (see https://docs.datastax.com/en/developer/java-driver/4.3/upgrade_guide/#load-balancing-policy[upgrade guide]).
+To customize load balancing policy, please use the parameter `loadBalancingPolicyClass` and provide own implementation
+of load balancing policy.
+
+Session impelentation class has changed from `com.datastax.driver.core.Session` to
+`com.datastax.oss.driver.api.core.CqlSession` (see https://docs.datastax.com/en/developer/java-driver/4.3/upgrade_guide/#session[upgrade guide]).
+This could have direct impact on the type of parameter `beanRef` if you are using `Provided Session reference`.
+
+There is a new parameter `datacenter` (with default value `datacenter1`). Data center has to be defined for the proper
+behavior of the default load balancing policy.
+
+
=== Camel Karaf
The following features has been removed due they become not compatible with OSGi: `camel-couchbase`.
diff --git a/parent/pom.xml b/parent/pom.xml
index c4b1fcb..0dec491 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -95,10 +95,10 @@
<c3p0-version>0.9.5.5</c3p0-version>
<caffeine-version>2.8.5</caffeine-version>
<californium-version>2.2.3</californium-version>
- <cassandra-driver-version>3.7.2</cassandra-driver-version>
- <cassandra-driver-guava-version>19.0</cassandra-driver-guava-version>
- <cassandra-version>3.7</cassandra-version>
- <cassandra-unit-version>3.1.3.2</cassandra-unit-version>
+ <cassandra-driver-version>4.7.2</cassandra-driver-version>
+ <cassandra-driver-guava-version>25.1-jre</cassandra-driver-guava-version>
+ <cassandra-version>3.11.5</cassandra-version>
+ <cassandra-unit-version>4.3.1.0</cassandra-unit-version>
<cdi-api-2.0-version>2.0</cdi-api-2.0-version>
<jta-api-1.2-version>1.2</jta-api-1.2-version>
<cglib-version>3.2.12</cglib-version>
@@ -2630,8 +2630,13 @@
<version>${cassandra-version}</version>
</dependency>
<dependency>
- <groupId>com.datastax.cassandra</groupId>
- <artifactId>cassandra-driver-core</artifactId>
+ <groupId>com.datastax.oss</groupId>
+ <artifactId>java-driver-core</artifactId>
+ <version>${cassandra-driver-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.datastax.oss</groupId>
+ <artifactId>java-driver-query-builder</artifactId>
<version>${cassandra-driver-version}</version>
</dependency>
<dependency>