remove mysql dep + add a configuration option for cassandra consistency

- use cassandra.consistency in properties file
- this fixes the problem where we have only one cassandra machine (hence quorum is not achievable and raises an error)

Author: derlin

pom.xml

@@ -12,13 +12,12 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <flink.version>1.10.0</flink.version>
+        <flink.version>1.10.2</flink.version>
         <java.version>1.8</java.version>
         <scala.binary.version>2.11</scala.binary.version>
         <maven.compiler.source>${java.version}</maven.compiler.source>
         <maven.compiler.target>${java.version}</maven.compiler.target>
 
-        <mysql.version>6.0.2</mysql.version>
         <slf4j.version>1.7.21</slf4j.version>
         <cassandra.version>3.9.0</cassandra.version>
     </properties>
@@ -102,13 +101,6 @@
 
         <!-- Other, regular dependencies -->
 
-        <!--mysql-->
-        <dependency>
-            <groupId>mysql</groupId>
-            <artifactId>mysql-connector-java</artifactId>
-            <version>${mysql.version}</version>
-        </dependency>
-
         <!--gson-->
         <dependency>
             <groupId>com.google.code.gson</groupId>

src/main/java/ch/derlin/bbdata/flink/Configs.java

@@ -1,5 +1,6 @@
 package ch.derlin.bbdata.flink;
 
+import com.datastax.driver.core.ConsistencyLevel;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
@@ -70,6 +71,12 @@ public class Configs {
             .noDefaultValue()
             .withDescription("List of entrypoints (IP, host) to connect to Cassandra, separated by semi-colons (;).");
 
+    public static final ConfigOption<String> configConsistency = ConfigOptions
+            .key("cassandra.consistency")
+            .stringType()
+            .defaultValue("QUORUM")
+            .withDescription("Cassandra consistency level. If you have a test cluster of one host, set it to ONE.");
+
     // =============== getters
 
     public static long readGranularity(Configuration config) {
@@ -123,10 +130,21 @@ public static List<String> readCassandraEntrypoints(Configuration config) {
         List<String> entrypoints = config.get(configEntrypoints);
         if (entrypoints.size() == 0)
             throw new RuntimeException(String.format(
-                    "Missing Cassandra entrypoing '%s' in config.", configEntrypoints.key()));
+                    "Missing Cassandra entrypoint '%s' in config.", configEntrypoints.key()));
         return entrypoints;
     }
 
+    public static ConsistencyLevel readCassandraConsistency(Configuration config) {
+        String consistency = config.get(configConsistency);
+        try {
+            return ConsistencyLevel.valueOf(consistency);
+        } catch (Exception e) {
+            throw new RuntimeException((String.format(
+                    "Wrong consistency value provided: '%s'. " +
+                            "See com.datastax.driver.core.ConsistencyLevel for possible values.", consistency)));
+        }
+    }
+
 
     public static void checkConfig(Configuration config) {
         // windows

src/main/java/ch/derlin/bbdata/flink/pojo/AggregationRecord.java

@@ -19,11 +19,7 @@
  *
  * @author Lucy Linder <[email protected]>
  */
-@Table(keyspace = "bbdata2", name = "aggregations",
-        readConsistency = "QUORUM",
-        writeConsistency = "QUORUM",
-        caseSensitiveKeyspace = false,
-        caseSensitiveTable = false)
+@Table(keyspace = "bbdata2", name = "aggregations")
 public class AggregationRecord {
 
     @PartitionKey(0)

src/main/java/ch/derlin/bbdata/flink/sinks/CassandraSink.java

@@ -6,6 +6,7 @@
 import ch.derlin.bbdata.flink.pojo.AggregationRecord;
 import ch.derlin.bbdata.flink.utils.DateUtil;
 import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.QueryOptions;
 import com.datastax.driver.core.Session;
 import com.datastax.driver.core.policies.DefaultRetryPolicy;
 import com.datastax.driver.mapping.Mapper;
@@ -16,6 +17,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static com.datastax.driver.mapping.Mapper.Option.consistencyLevel;
+
 /**
  * date: 10.03.17
  *
@@ -48,18 +51,26 @@ public void open(Configuration parameters) throws Exception {
                     // TODO does it improve performances ?
                     ;
             for (String address : Configs.readCassandraEntrypoints(config)) {
+                LOG.info("adding cassandra contact point: " + address.trim());
                 builder.addContactPoint(address.trim());
             }
 
+            // set consistency
+            QueryOptions queryOptions = new QueryOptions();
+            queryOptions.setConsistencyLevel(Configs.readCassandraConsistency(config));
+
             // the keyspace and tables are declared in the AggregationRecord class
             // so here, just create a connection
-            cluster = builder.build();
+            cluster = builder
+                    .withQueryOptions(queryOptions)
+                    .build();
             session = cluster.connect();
 
             // the mapper will do the mapping between cassandra records and AggregationRecord objects
             MappingManager manager = new MappingManager(session);
             mapper = manager.mapper(AggregationRecord.class);
 
+
         } catch (Exception e) {
             LOG.error("setup/open failed", e);
             throw e;

src/main/resources/config.properties-template

@@ -3,7 +3,7 @@
 # =====
 # configuration for communicating with Kafka
 #
-# the brokers address:port, comma-separated
+# the brokers address:port, separated by ;
 kafka.brokers=localhost:6667
 # the source topic, with the augmented values
 kafka.augmentation=bbdata2-augmented
@@ -17,8 +17,11 @@ kafka.consumer.group=flink-aggregation-01
 # configuration for saving aggregations to cassandra.
 # the cassandra instance should have a table bbdata2.aggregations table !
 #
-# the cassandra entrypoints, a list of ip addresses separated by comma
+# the cassandra entrypoints, a list of ip addresses separated by ;
 cassandra.entrypoints=127.0.0.1
+# the cassandra consistency level to use. See enum com.datastax.driver.core.ConsistencyLevel
+# NOTE: if you have only one cassandra instance, ensure you set it to ONE !
+cassandra.consistency=QUORUM
 
 # Flink checkpoints
 # =================

src/test/java/ch/derlin/bbdata/flink/TestConfigs.java

@@ -119,4 +119,17 @@ public void testCassandraEntrypoints(){
         assertEquals(2, Configs.readCassandraEntrypoints(config).size());
 
     }
+
+
+    @Test
+    public void testCassandraConsistency(){
+        Configuration config = new Configuration();
+        assertDoesNotThrow(() -> Configs.readCassandraConsistency(config));
+
+        config.setString(Configs.configConsistency.key(), "ONE");
+        assertDoesNotThrow(() -> Configs.readCassandraConsistency(config));
+
+        config.setString(Configs.configConsistency.key(), "LALA");
+        assertThrows(Exception.class, () -> Configs.readCassandraConsistency(config));
+    }
 }