Resolves Issue #57 - Add Bulk{Write, Read} to NDBench. (#64)

* Issue #57 * Issue #57 - addressed review comments
Netflix · Feb 8, 2018 · a6612fe · a6612fe
1 parent f4d3783
commit a6612fe
Show file tree

Hide file tree

Showing 24 changed files with 479 additions and 91 deletions.
diff --git a/ndbench-api/src/main/java/com/netflix/ndbench/api/plugin/NdBenchAbstractClient.java b/ndbench-api/src/main/java/com/netflix/ndbench/api/plugin/NdBenchAbstractClient.java
@@ -17,6 +17,8 @@
 
 package com.netflix.ndbench.api.plugin;
 
+import java.util.List;
+
 /**
  * Defines default methods that provide a hook point for Auto-tuning, more information for which may be found
  * <a href="https://github.com/Netflix/ndbench/wiki/Configuration">here</a>.
@@ -27,6 +29,8 @@
  * to the type that heretofore has been returned by all clients implementing the writeSingle method (namely: String.)
  *
  * @param <W> - the type of the result returned by {@link #writeSingle}
+ *
+ * @author vchella, pencal
  */
 public interface NdBenchAbstractClient<W> {
 
@@ -45,6 +49,13 @@ public interface NdBenchAbstractClient<W> {
      */
     String readSingle(final String key) throws Exception;
 
+    /**
+     * Perform a bulk read operation given the list of keys
+     *
+     * @return
+     * @throws Exception
+     */
+    List<String> readBulk(final List<String> keys) throws Exception;
 
     /**
      * Perform a single write operation
@@ -54,6 +65,13 @@ public interface NdBenchAbstractClient<W> {
      */
     W writeSingle(final String key) throws Exception;
 
+    /**
+     * Perform bulk write operation given the list of keys
+     *
+     * @param keys
+     * @return
+     */
+    List<W> writeBulk(final List<String> keys) throws Exception;
 
     /**
      * shutdown the client
@@ -94,14 +112,14 @@ public interface NdBenchAbstractClient<W> {
      * @return - the new suggested rate limit -- ignored by driver if less-than-or-equal-to 0.
      */
 
-    default Double autoTuneWriteRateLimit(Double currentRateLimit, W event, NdBenchMonitor runStats) {
+    default Double autoTuneWriteRateLimit(Double currentRateLimit, List<W> event, NdBenchMonitor runStats) {
         return -1D;
     }
 
     /**
      * See documentation for {@link #autoTuneWriteRateLimit}
      */
-    default double autoTuneReadRateLimit(double currentRateLimit, W event, NdBenchMonitor runStats) {
+    default double autoTuneReadRateLimit(double currentRateLimit, List<W> event, NdBenchMonitor runStats) {
         throw new UnsupportedOperationException("not yet implemented");
     }
 }
diff --git a/ndbench-api/src/main/java/com/netflix/ndbench/api/plugin/NdBenchBaseClient.java b/ndbench-api/src/main/java/com/netflix/ndbench/api/plugin/NdBenchBaseClient.java
@@ -18,19 +18,31 @@
 package com.netflix.ndbench.api.plugin;
 
 
+import java.util.List;
+
 /**
- * @author vchella
+ * @author vchella, pencal
  */
 public abstract class NdBenchBaseClient implements NdBenchClient {
 
     @Override
-    public String readSingle(String key) throws Exception {
+    public String readSingle(final String key) throws Exception {
         return null;
     }
 
 
     @Override
-    public String writeSingle(String key) throws Exception {
+    public String writeSingle(final String key) throws Exception {
+        return null;
+    }
+
+    @Override
+    public List<String> readBulk(final List<String> keys) throws Exception {
+        return null;
+    }
+
+    @Override
+    public List<String> writeBulk(final List<String> keys) throws Exception {
         return null;
     }
 

diff --git a/ndbench-api/src/main/java/com/netflix/ndbench/api/plugin/NdBenchClient.java b/ndbench-api/src/main/java/com/netflix/ndbench/api/plugin/NdBenchClient.java
@@ -17,8 +17,10 @@
 
 package com.netflix.ndbench.api.plugin;
 
+import java.util.List;
+
 /**
- * @author vchella
+ * @author vchella, pencal
  */
  public interface NdBenchClient extends NdBenchAbstractClient<String> {
 
@@ -35,16 +37,29 @@ public interface NdBenchClient extends NdBenchAbstractClient<String> {
 	 */
 	 String readSingle(final String key) throws Exception;
 
-
 	/**
 	 * Perform a single write operation
 	 * @return
 	 * @throws Exception
 	 */
 	 String writeSingle(final String key) throws Exception;
 
+    /**
+     * Perform a bulk read operation
+     * @return
+     * @throws Exception
+     */
+    List<String> readBulk(final List<String> keys) throws Exception;
 
     /**
+	 * Perform a bulk write operation
+	 * @return
+	 * @throws Exception
+	 */
+	List<String> writeBulk(final List<String> keys) throws Exception;
+
+
+	/**
      * shutdown the client
      */
      void shutdown() throws Exception;

diff --git a/ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassAstyanaxPlugin.java b/ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassAstyanaxPlugin.java
@@ -40,6 +40,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import static com.netflix.ndbench.api.plugin.common.NdBenchConstants.PROP_NAMESPACE;
 
 
@@ -181,6 +184,24 @@ public void shutdown() throws Exception {
 
     }
 
+    /**
+     * Perform a bulk read operation
+     * @return a list of response codes
+     * @throws Exception
+     */
+    public List<String> readBulk(final List<String> keys) throws Exception {
+        throw new UnsupportedOperationException("bulk operation is not supported");
+    }
+
+    /**
+     * Perform a bulk write operation
+     * @return a list of response codes
+     * @throws Exception
+     */
+    public List<String> writeBulk(final List<String> keys) throws Exception {
+        throw new UnsupportedOperationException("bulk operation is not supported");
+    }
+
     /**
      * Get connection info
      */

diff --git a/ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassJavaDriverBatch.java b/ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassJavaDriverBatch.java
@@ -11,6 +11,7 @@
 import com.netflix.ndbench.api.plugin.common.NdBenchConstants;
 
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
 /**
@@ -125,6 +126,24 @@ public String writeSingle(String key) throws Exception {
         return ResultOK;
     }
 
+    /**
+     * Perform a bulk read operation
+     * @return a list of response codes
+     * @throws Exception
+     */
+    public List<String> readBulk(final List<String> keys) throws Exception {
+        throw new UnsupportedOperationException("bulk operation is not supported");
+    }
+
+    /**
+     * Perform a bulk write operation
+     * @return a list of response codes
+     * @throws Exception
+     */
+    public List<String> writeBulk(final List<String> keys) throws Exception {
+        throw new UnsupportedOperationException("bulk operation is not supported");
+    }
+
     private BoundStatement getBStmtTable1(String key) {
         BoundStatement bStmt = writePstmt.bind();
         bStmt.setString("cyclist_name", key);

diff --git a/...nch-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassJavaDriverGeneric.java b/...nch-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassJavaDriverGeneric.java
@@ -8,6 +8,7 @@
 import com.netflix.ndbench.api.plugin.annotations.NdBenchClientPlugin;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
 import java.util.List;
 
 
@@ -66,6 +67,24 @@ public String writeSingle(String key) throws Exception {
         return ResultOK;
     }
 
+    /**
+     * Perform a bulk read operation
+     * @return a list of response codes
+     * @throws Exception
+     */
+    public List<String> readBulk(final List<String> keys) throws Exception {
+        throw new UnsupportedOperationException("bulk operation is not supported");
+    }
+
+    /**
+     * Perform a bulk write operation
+     * @return a list of response codes
+     * @throws Exception
+     */
+    public List<String> writeBulk(final List<String> keys) throws Exception {
+        throw new UnsupportedOperationException("bulk operation is not supported");
+    }
+
     @Override
     void upsertKeyspace(Session session) {
        upsertGenereicKeyspace();

diff --git a/ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassJavaDriverPlugin.java b/ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassJavaDriverPlugin.java
diff --git a/...s/src/main/java/com/netflix/ndbench/plugin/elassandra/ElassandraCassJavaDriverPlugin.java b/...s/src/main/java/com/netflix/ndbench/plugin/elassandra/ElassandraCassJavaDriverPlugin.java
@@ -23,6 +23,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
@@ -128,6 +129,24 @@ public String writeSingle(String key) throws Exception {
         return ResultOK;
     }
 
+    /**
+     * Perform a bulk read operation
+     * @return a list of response codes
+     * @throws Exception
+     */
+    public List<String> readBulk(final List<String> keys) throws Exception {
+        throw new UnsupportedOperationException("bulk operation is not supported");
+    }
+
+    /**
+     * Perform a bulk write operation
+     * @return a list of response codes
+     * @throws Exception
+     */
+    public List<String> writeBulk(final List<String> keys) throws Exception {
+        throw new UnsupportedOperationException("bulk operation is not supported");
+    }
+
     /**
      * shutdown the client
      */