Rate Limting

Added Rate Limiting Sample

Author: michaelraney

.gitignore

@@ -1,2 +1,4 @@
-
+.idea
+*.iml
 .DS_Store
+/target

README.md

@@ -20,6 +20,45 @@ basic.request.default-idempotence = true
 }
 ```
 
+## Throttling - RateLimiting
+Retries maintain level of availability when receiving short burst of traffic, acute failure, or loss of connection, but sustained retries can further destabilize systems resulting in cascading failure. If you are using retries to limit traffic then you may want to consider a rate limiter.  As reties continue to occur at a steady rate they increasingly add to the overall traffic sent to the database.  When facing this scenario you should introduce rate-limiting. Rate limiters provide what is known as back pressure. You can achieve this by leveraging the Java Driver's Throttler Extension point.  There are a few rate-limiters provided native with the driver, but in this repository we will provide some sample limiters that are designed for Amazon Keyspaces serverless capacity and service quotas. 
+
+### AmazonKeyspacesFixedRateThrottler
+This is a request throttler that limits the rate of requests per second. It can be used to match the current table provisioned rate if you have predictable or average capacity per request.
+The limit is configurable through the client driver configuration. 
+The rate of request is controlled by Guava SmoothBursty Ratelimiter that allows two minutes of capacity to aggregate if not used. 
+The second limiter is dynamically configured based on the number of connections defined in the pool setting that limits overall throughput during burst behavior. 
+The limiter will control the number of cql request per second but expects the table to have proper capacity defined to achieve utilization. 
+This is a blocking implementation, but has a configurable timeout. 
+    
+The most well-known use-case for this type of rate limiter is bulk loading data at consistent rates or batch processing. In the image below we are able to utilize 100 percent of the table's capacity without error by fixing the request rate to the table write capacity. The workload gradually stepped up throughput level all the way up to 3x the request rate of the table's provisioned capacity. Additionally, since the rate limiter allows for bursting, we are able to use burst capacity of the table that can accrue when provisioned capacity is not fully utilized. The fixed are limiter can also be used with On-Demand Capacity Tables to fit within current account specific table quotas.  
+![Rate Limiting](/static/images/RateLimiting.png) 
+ 
+To activate this throttler, modify the {@code advanced.throttler} section in the driver configuration, for example:
+     
+   ```
+      datastax-java-driver {
+         advanced.throttler = {
+                class = com.aws.ssa.keyspaces.throttler.AmazonKeyspacesFixedRateThrottler
+                max-requests-per-second = 1000
+                endpoint-type = VPC
+                register-timeout = 1 seconds
+          }
+      }  
+```
+
+
+
+* `max-requests-per-second` : controls the request rate. Blocks until available permits or timeout it reached
+* `endpoint-type` : if you are connected to the VPC endpoint or the public endpoint. Used to determine throughput based on the number of connections specified in:`advanced.connection.pool.local.size`
+* `register-timeout` timeout waiting for permits. Should be less than or equal to `basic.request.timeout'
+
+
+# Build this project
+To build and use this library execute the following mvn command and place on the classpath of your application. 
+```
+mvn clean package
+```
 # Security
 
 See [CONTRIBUTING](CONTRIBUTING.md#security-issue-notifications) for more information.

pom.xml

@@ -4,25 +4,22 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
 
-    <groupId>org.example</groupId>
+    <groupId>com.aws.ssa.keyspaces</groupId>
     <artifactId>amazon-keyspaces-helpers</artifactId>
     <version>1.0-SNAPSHOT</version>
 
     <properties>
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-
         <aws.java.sdk.version>2.13.7</aws.java.sdk.version>
-
     </properties>
 
 
     <dependencies>
-
         <dependency>
             <groupId>com.datastax.oss</groupId>
             <artifactId>java-driver-core</artifactId>
-            <version>4.10.0</version>
+            <version>4.2.0</version>
             <scope>provided</scope>
         </dependency>
         <dependency>
@@ -43,16 +40,9 @@
             <version>1.2.3</version>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>io.dropwizard.metrics</groupId>
-            <artifactId>metrics-core</artifactId>
-            <version>4.1.0</version>
-            <scope>test</scope>
-        </dependency>
     </dependencies>
 
     <build>
-
         <plugins>
             <plugin>
                 <artifactId>maven-assembly-plugin</artifactId>

src/main/java/com/aws/ssa/keyspaces/core/EndpointType.java

@@ -0,0 +1,6 @@
+package com.aws.ssa.keyspaces.core;
+
+public enum EndpointType {
+    VPC,
+    PUBLIC;
+}

src/main/java/com/aws/ssa/keyspaces/retry/AmazonKeyspacesRetryPolicy.java

@@ -105,13 +105,10 @@ public RetryDecision onReadTimeout(
             boolean dataPresent,
             int retryCount) {
 
-
         RetryDecision decision = determineRetryDecision(retryCount);
 
-        if (LOG.isTraceEnabled() && (decision == RetryDecision.RETRY_SAME || decision == RetryDecision.RETHROW)) {
-            LOG.trace(RETRYING_ON_READ_TIMEOUT, logPrefix, cl, blockFor, received, false, retryCount);
-        }
-
+        LOG.trace(RETRYING_ON_READ_TIMEOUT, logPrefix, cl, blockFor, received, false, retryCount);
+        
         return decision;
 
     }
@@ -133,12 +130,10 @@ public RetryDecision onWriteTimeout(
             int received,
             int retryCount) {
 
-
         RetryDecision decision = determineRetryDecision(retryCount);
 
-        if (LOG.isTraceEnabled() && (decision == RetryDecision.RETRY_SAME || decision == RetryDecision.RETHROW)) {
-            LOG.trace(RETRYING_ON_WRITE_TIMEOUT, logPrefix, cl, blockFor, received, false, retryCount);
-        }
+        LOG.trace(RETRYING_ON_WRITE_TIMEOUT, logPrefix, cl, blockFor, received, false, retryCount);
+
 
         return decision;
     }
@@ -160,9 +155,8 @@ public RetryDecision onUnavailable(
 
         RetryDecision decision = determineRetryDecision(retryCount);
 
-        if (LOG.isTraceEnabled() && (decision == RetryDecision.RETRY_SAME || decision == RetryDecision.RETHROW)) {
-            LOG.trace(RETRYING_ON_UNAVAILABLE, logPrefix, cl, required, alive, retryCount);
-        }
+        LOG.trace(RETRYING_ON_UNAVAILABLE, logPrefix, cl, required, alive, retryCount);
+
 
         return decision;
     }
@@ -175,11 +169,11 @@ public RetryDecision onUnavailable(
     @Override
     public RetryDecision onRequestAborted(
             @NonNull Request request, @NonNull Throwable error, int retryCount) {
+
         RetryDecision decision = determineRetryDecision(retryCount);
 
-        if (LOG.isTraceEnabled() && (decision == RetryDecision.RETRY_SAME || decision == RetryDecision.RETHROW)) {
-            LOG.trace(RETRYING_ON_ABORTED, logPrefix, retryCount, error);
-        }
+        LOG.trace(RETRYING_ON_ABORTED, logPrefix, retryCount, error);
+
 
         return decision;
     }
@@ -193,12 +187,9 @@ public RetryDecision onRequestAborted(
     public RetryDecision onErrorResponse(
             @NonNull Request request, @NonNull CoordinatorException error, int retryCount) {
 
-
         RetryDecision decision = determineRetryDecision(retryCount);
 
-        if (LOG.isTraceEnabled() && (decision == RetryDecision.RETRY_SAME || decision == RetryDecision.RETHROW)) {
-            LOG.trace(RETRYING_ON_ERROR, logPrefix, retryCount, error);
-        }
+        LOG.trace(RETRYING_ON_ERROR, logPrefix, retryCount, error);
 
         return decision;
     }