CassJavaDriverGeneric: add options to disable use of batch writes

The use of BATCH in CassJavaDriverGeneric allows the driver to write
multiple rows in a single call to `writeSingle`. Its sometimes
desirable, however, to have `writeSingle` to truly write a single row
so that BATCH is not used.

This can cause the row count validation to fail since not every row is
written to every partition. Ad additional configuration has been added
to disable this valdiation.

ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassJavaDriverGeneric.java

@@ -67,7 +67,7 @@ public String readSingle(String key) throws Exception {
         if (!result.isEmpty())
         {
             nRows = result.size();
-            if (nRows < (config.getRowsPerPartition()))
+            if (config.getValidateRowsPerPartition() && nRows < (config.getRowsPerPartition()))
             {
                 throw new Exception("Num rows returned not ok " + nRows);
             }
@@ -99,14 +99,19 @@ public String writeSingle(String key)
     {
         if(config.getRowsPerPartition() > 1)
         {
-            BatchStatement batch = new BatchStatement(BatchStatement.Type.UNLOGGED);
-            batch.setConsistencyLevel(ConsistencyLevel.valueOf(config.getWriteConsistencyLevel()));
-            for (int i = 0; i < config.getRowsPerPartition(); i++)
-            {
-                batch.add(getWriteBStmt(key,i));
+            if (config.getUseBatchWrites()) {
+                BatchStatement batch = new BatchStatement(BatchStatement.Type.UNLOGGED);
+                batch.setConsistencyLevel(ConsistencyLevel.valueOf(config.getWriteConsistencyLevel()));
+                for (int i = 0; i < config.getRowsPerPartition(); i++) {
+                    batch.add(getWriteBStmt(key, i));
+                }
+                session.execute(batch);
+                batch.clear();
+            } else {
+                session.execute(getWriteBStmt(key, dataGenerator.getRandomInteger() % config.getRowsPerPartition())
+                        .setConsistencyLevel(ConsistencyLevel.valueOf(config.getWriteConsistencyLevel())));
+
             }
-            session.execute(batch);
-            batch.clear();
         }
         else
         {

ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/configs/CassandraGenericConfiguration.java

@@ -23,6 +23,13 @@
 public interface CassandraGenericConfiguration extends CassandraConfigurationBase {
     @DefaultValue("2")
     Integer getRowsPerPartition();
+
     @DefaultValue("5")
     Integer getColsPerRow();
+
+    @DefaultValue("false")
+    Boolean getUseBatchWrites();
+
+    @DefaultValue("false")
+    Boolean getValidateRowsPerPartition();
 }