Method flush

src/HBaseClient.java:801–866

Flushes to HBase any buffered client-side write operation. @return A Deferred, whose callback chain will be invoked when everything that was buffered at the time of the call has been flushed. Note that this doesn't guarantee that ALL outstanding RPCs have completed. This does

()

   * it does really is it sends any buffered RPCs to HBase.
   */
  public Deferred<Object> flush() {
    {
      // If some RPCs are waiting for -ROOT- to be discovered, we too must wait
      // because some of those RPCs could be edits that we must wait on.
      final Deferred<Object> d = zkclient.getDeferredRootIfBeingLookedUp();
      if (d != null) {
        LOG.debug("Flush needs to wait on {} to come back",
          has_root ? (split_meta ? new String(HBASE98_ROOT_REGION) : new String(ROOT)) 
              : (split_meta ? new String(HBASE96_META) : new String(META)));
        final class RetryFlush implements Callback<Object, Object> {
          public Object call(final Object arg) {
            LOG.debug("Flush retrying after {} came back",
                has_root ? (split_meta ? new String(HBASE98_ROOT_REGION) : new String(ROOT)) 
                    : (split_meta ? new String(HBASE96_META) : new String(META)));
            return flush();
          }
          public String toString() {
            return "retry flush";
          }
        }
        return d.addBoth(new RetryFlush());
      }
    }

    num_flushes.increment();
    final boolean need_sync;
    {
      final LoadingCache<BufferedIncrement, BufferedIncrement.Amount> buf =
        increment_buffer;  // Single volatile-read.
      if (buf != null && !buf.asMap().isEmpty()) {
        flushBufferedIncrements(buf);
        need_sync = true;

      } else {
        final LoadingCache<BufferedMultiColumnIncrement, BufferedMultiColumnIncrement.Amounts> multiColumnBuf =
          multi_column_increment_buffer;  // Single volatile-read.
        if (multiColumnBuf != null && !multiColumnBuf.asMap().isEmpty()) {
          flushBufferedMultiColumnIncrements(multiColumnBuf);
          need_sync = true;

        } else {
          need_sync = false;
        }
      }
    }
    final ArrayList<Deferred<Object>> d =
      new ArrayList<Deferred<Object>>(client2regions.size()
                                      + got_nsre.size() * 8);
    // Bear in mind that we're traversing a ConcurrentHashMap, so we may get
    // clients that have been removed from the map since we started iterating.
    for (final RegionClient client : client2regions.keySet()) {
      d.add(need_sync ? client.sync() : client.flush());
    }
    for (final ArrayList<HBaseRpc> nsred : got_nsre.values()) {
      synchronized (nsred) {
        for (final HBaseRpc rpc : nsred) {
          if (rpc instanceof HBaseRpc.IsEdit) {
            d.add(rpc.getDeferred());