title: What is the best way to get backpressure for Cassandra Writes?
body: <p>I have a service that consumes messages off of a queue at a rate that I control.  I do some processing and then attempt to write to a Cassandra cluster via the Datastax Java client.  I have setup my Cassandra cluster with <code>maxRequestsPerConnection</code> and <code>maxConnectionsPerHost</code>.  However, in testing I have found that when I have reached <code>maxConnectionsPerHost</code> and <code>maxRequestsPerConnection</code> calls to <code>session.executeAsync</code> don't block.</p>
  
  <p>What I am doing right now is using a <code>new Semaphore(maxConnectionsPerHost * maxRequestsPerConnection)</code> and incrementing it before every async request and decrementing it when the future returned by <code>executeAsync</code> completes.  This works well enough, but it seems redundant since the driver is already tracking requests and connections internally.</p>
  
  <p>Has anyone come up with a better solution to this problem?</p>
  
  <p>One caveat: I would like a request to be considered outstanding until it has completed.  This <em>includes retries</em>!  The situation where I am getting retryable failures from the cluster (such as timeouts waiting for consistency) is primary situation where I want to backpressure and stop consuming messages from the queue.</p>
  
  <p>Problem:</p>
  
  <pre><code>// the rate at which I consume messages depends on how fast this method returns
  processMessage(message) {
      // this appears to return immediately even if I have exhausted connections/requests
      session.executeAsync(preparedStatement.bind(...));
  }
  </code></pre>
  
  <p>Current solution:</p>
  
  <pre><code>constructor() {
      this.concurrentRequestsSemaphore = new Semaphore(maxConnectionsPerHost * maxRequestsPerConnection);
  }
  
  processMessage(message) {
      ResultSetFuture resultSetFuture = session.executeAsync(preparedStatement.bind(...));
      CompletableFuture&lt;ResultSet&gt; future = completableFromListenable(resultSetFuture);
      concurrentRequestsSemaphore.acquireUninterruptibly();
      future.whenComplete((result, exception) -&gt; concurrentRequests.release());
  }
  </code></pre>
  
  <p>Also, can anyone see any obvious problems with this solution?</p>