code: public void startNewBatchForCluster(String cluster) {
          // Only start a new batch if the current one is non-empty
          PlanBatch batch = _clusterTails.get(cluster);
          if (batch != null && !batch.getItems().isEmpty()) {
              PlanBatch newBatch = new PlanBatch();
              batch.setNextBatch(newBatch);
              _clusterTails.put(cluster, newBatch);
          }
      }
docs: Starts a new batch for a cluster.  Future calls to {@link #addTokenRangeToCurrentBatchForCluster(String, String, java.util.Collection)}
  are added to the batch started by this call for the corresponding cluster.