tephra-hbase-compat-2.0-base/src/main/java/org/apache/tephra/hbase/TransactionAwareHTable.java [72:498]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class TransactionAwareHTable extends AbstractTransactionAwareTable
    implements Table, TransactionAware {

  private static final Logger LOG = LoggerFactory.getLogger(TransactionAwareHTable.class);
  private final Table hTable;

  /**
   * Create a transactional aware instance of the passed HTable
   * @param hTable underlying HBase table to use
   */
  public TransactionAwareHTable(Table hTable) {
    this(hTable, false);
  }

  /**
   * Create a transactional aware instance of the passed HTable
   * @param hTable underlying HBase table to use
   * @param conflictLevel level of conflict detection to perform (defaults to {@code COLUMN})
   */
  public TransactionAwareHTable(Table hTable, TxConstants.ConflictDetection conflictLevel) {
    this(hTable, conflictLevel, false);
  }

  /**
   * Create a transactional aware instance of the passed HTable, with the option of allowing
   * non-transactional operations.
   * @param hTable underlying HBase table to use
   * @param allowNonTransactional if true, additional operations (checkAndPut, increment,
   *          checkAndDelete) will be available, though non-transactional
   */
  public TransactionAwareHTable(Table hTable, boolean allowNonTransactional) {
    this(hTable, TxConstants.ConflictDetection.COLUMN, allowNonTransactional);
  }

  /**
   * Create a transactional aware instance of the passed HTable, with the option of allowing
   * non-transactional operations.
   * @param hTable underlying HBase table to use
   * @param conflictLevel level of conflict detection to perform (defaults to {@code COLUMN})
   * @param allowNonTransactional if true, additional operations (checkAndPut, increment,
   *          checkAndDelete) will be available, though non-transactional
   */
  public TransactionAwareHTable(Table hTable, TxConstants.ConflictDetection conflictLevel,
      boolean allowNonTransactional) {
    super(conflictLevel, allowNonTransactional, 
        hTable.getConfiguration().getBoolean(
            TxConstants.TX_PRE_014_CHANGESET_KEY,
            TxConstants.DEFAULT_TX_PRE_014_CHANGESET_KEY));
    this.hTable = hTable;
  }

  /* AbstractTransactionAwareTable implementation */

  @Override
  protected byte[] getTableKey() {
    return hTable.getName().getName();
  }

  @Override
  protected boolean doCommit() throws IOException {
    return true;
  }

  @Override
  protected boolean doRollback() throws Exception {
    try {
      // pre-size arraylist of deletes
      int size = 0;
      for (Set<ActionChange> cs : changeSets.values()) {
        size += cs.size();
      }
      List<Delete> rollbackDeletes = new ArrayList<>(size);
      for (Map.Entry<Long, Set<ActionChange>> entry : changeSets.entrySet()) {
        long transactionTimestamp = entry.getKey();
        for (ActionChange change : entry.getValue()) {
          byte[] row = change.getRow();
          byte[] family = change.getFamily();
          byte[] qualifier = change.getQualifier();
          Delete rollbackDelete = new Delete(row);
          makeRollbackOperation(rollbackDelete);
          switch (conflictLevel) {
          case ROW:
          case NONE:
            // issue family delete for the tx write pointer
            rollbackDelete.addFamilyVersion(change.getFamily(), transactionTimestamp);
            break;
          case COLUMN:
            if (family != null && qualifier == null) {
              rollbackDelete.addFamilyVersion(family, transactionTimestamp);
            } else if (family != null && qualifier != null) {
              rollbackDelete.addColumn(family, qualifier, transactionTimestamp);
            }
            break;
          default:
            throw new IllegalStateException("Unknown conflict detection level: " + conflictLevel);
          }
          rollbackDeletes.add(rollbackDelete);
        }
      }
      hTable.delete(rollbackDeletes);
      return true;
    } finally {
      tx = null;
      changeSets.clear();
    }
  }

  /* HTableInterface implementation */

  @Override
  public TableName getName() {
    return hTable.getName();
  }

  @Override
  public Configuration getConfiguration() {
    return hTable.getConfiguration();
  }

  @Override
  public HTableDescriptor getTableDescriptor() throws IOException {
    return hTable.getTableDescriptor();
  }

  @Override
  public boolean exists(Get get) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    return hTable.exists(transactionalizeAction(get));
  }

  @Override
  public void batch(List<? extends Row> actions, Object[] results)
      throws IOException, InterruptedException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    hTable.batch(transactionalizeActions(actions), results);
  }

  @Override
  public <R> void batchCallback(List<? extends Row> actions, Object[] results,
      Batch.Callback<R> callback) throws IOException, InterruptedException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    hTable.batchCallback(transactionalizeActions(actions), results, callback);
  }

  @Override
  public Result get(Get get) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    return hTable.get(transactionalizeAction(get));
  }

  @Override
  public Result[] get(List<Get> gets) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    ArrayList<Get> transactionalizedGets = new ArrayList<>();
    for (Get get : gets) {
      transactionalizedGets.add(transactionalizeAction(get));
    }
    return hTable.get(transactionalizedGets);
  }

  @Override
  public ResultScanner getScanner(Scan scan) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    return hTable.getScanner(transactionalizeAction(scan));
  }

  @Override
  public ResultScanner getScanner(byte[] family) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    Scan scan = new Scan();
    scan.addFamily(family);
    return hTable.getScanner(transactionalizeAction(scan));
  }

  @Override
  public ResultScanner getScanner(byte[] family, byte[] qualifier) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    Scan scan = new Scan();
    scan.addColumn(family, qualifier);
    return hTable.getScanner(transactionalizeAction(scan));
  }

  @Override
  public void put(Put put) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    Put txPut = transactionalizeAction(put);
    hTable.put(txPut);
  }

  @Override
  public void put(List<Put> puts) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    List<Put> transactionalizedPuts = new ArrayList<>(puts.size());
    for (Put put : puts) {
      Put txPut = transactionalizeAction(put);
      transactionalizedPuts.add(txPut);
    }
    hTable.put(transactionalizedPuts);
  }

  @Override
  public void delete(Delete delete) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    hTable.put(transactionalizeAction(delete));
  }

  @Override
  public void delete(List<Delete> deletes) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    List<Put> transactionalizedDeletes = new ArrayList<>(deletes.size());
    for (Delete delete : deletes) {
      Put txDelete = transactionalizeAction(delete);
      transactionalizedDeletes.add(txDelete);
    }
    hTable.put(transactionalizedDeletes);
  }

  @Override
  public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier, byte[] value, Put put)
      throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndPut(row, family, qualifier, value, put);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier, byte[] value,
      Delete delete) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndDelete(row, family, qualifier, value, delete);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndDelete(byte[] bytes, byte[] bytes1, byte[] bytes2,
      CompareFilter.CompareOp compareOp, byte[] bytes3, Delete delete) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndDelete(bytes, bytes1, bytes2, compareOp, bytes3, delete);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndPut(byte[] bytes, byte[] bytes1, byte[] bytes2,
      CompareFilter.CompareOp compareOp, byte[] bytes3, Put put) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndPut(bytes, bytes1, bytes2, compareOp, bytes3, put);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndMutate(byte[] row, byte[] family, byte[] qualifier,
      CompareFilter.CompareOp compareOp, byte[] value, RowMutations rowMutations)
      throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndMutate(row, family, qualifier, compareOp, value, rowMutations);
    }

    throw new UnsupportedOperationException(
        "checkAndMutate operation is not supported transactionally");
  }

  @Override
  public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier, CompareOperator op,
      byte[] value, Put put) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndPut(row, family, qualifier, value, put);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }

  }

  @Override
  public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier, CompareOperator op,
      byte[] value, Delete delete) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndDelete(row, family, qualifier, op, value, delete);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndMutate(byte[] row, byte[] family, byte[] qualifier, CompareOperator op,
      byte[] value, RowMutations mutation) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndMutate(row, family, qualifier, op, value, mutation);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean[] existsAll(List<Get> gets) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    List<Get> transactionalizedGets = new ArrayList<>(gets.size());
    for (Get get : gets) {
      transactionalizedGets.add(transactionalizeAction(get));
    }
    return hTable.existsAll(transactionalizedGets);
  }

  @Override
  public void mutateRow(RowMutations rm) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    RowMutations transactionalMutations = new RowMutations(rm.getRow());
    for (Mutation mutation : rm.getMutations()) {
      if (mutation instanceof Put) {
        transactionalMutations.add(transactionalizeAction((Put) mutation));
      } else if (mutation instanceof Delete) {
        transactionalMutations.add(transactionalizeAction((Delete) mutation));
      }
    }
    hTable.mutateRow(transactionalMutations);
  }

  @Override
  public Result append(Append append) throws IOException {
    if (allowNonTransactional) {
      return hTable.append(append);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public Result increment(Increment increment) throws IOException {
    if (allowNonTransactional) {
      return hTable.increment(increment);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public long incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount)
      throws IOException {
    if (allowNonTransactional) {
      return hTable.incrementColumnValue(row, family, qualifier, amount);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public long incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount,
      Durability durability) throws IOException {
    if (allowNonTransactional) {
      return hTable.incrementColumnValue(row, family, qualifier, amount, durability);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public void close() throws IOException {
    hTable.close();
  }

  @Override
  public CoprocessorRpcChannel coprocessorService(byte[] row) {
    return hTable.coprocessorService(row);
  }

  @Override
  public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> service, byte[] startKey,
      byte[] endKey, Batch.Call<T, R> callable) throws ServiceException, Throwable {
    return hTable.coprocessorService(service, startKey, endKey, callable);
  }

  @Override
  public <T extends Service, R> void coprocessorService(Class<T> service, byte[] startKey,
      byte[] endKey, Batch.Call<T, R> callable, Batch.Callback<R> callback)
      throws ServiceException, Throwable {
    hTable.coprocessorService(service, startKey, endKey, callable, callback);
  }

  @Override
  public <R extends Message> Map<byte[], R> batchCoprocessorService(
      MethodDescriptor methodDescriptor, Message request, byte[] startKey, byte[] endKey,
      R responsePrototype) throws ServiceException, Throwable {
    return hTable.batchCoprocessorService(methodDescriptor, request, startKey, endKey,
      responsePrototype);
  }

  @Override
  public <R extends Message> void batchCoprocessorService(MethodDescriptor methodDescriptor,
      Message request, byte[] startKey, byte[] endKey, R responsePrototype, Callback<R> callback)
      throws ServiceException, Throwable {
    hTable.batchCoprocessorService(methodDescriptor, request, startKey, endKey, responsePrototype,
      callback);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



tephra-hbase-compat-2.3/src/main/java/org/apache/tephra/hbase/TransactionAwareHTable.java [73:499]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class TransactionAwareHTable extends AbstractTransactionAwareTable
    implements Table, TransactionAware {

  private static final Logger LOG = LoggerFactory.getLogger(TransactionAwareHTable.class);
  private final Table hTable;

  /**
   * Create a transactional aware instance of the passed HTable
   * @param hTable underlying HBase table to use
   */
  public TransactionAwareHTable(Table hTable) {
    this(hTable, false);
  }

  /**
   * Create a transactional aware instance of the passed HTable
   * @param hTable underlying HBase table to use
   * @param conflictLevel level of conflict detection to perform (defaults to {@code COLUMN})
   */
  public TransactionAwareHTable(Table hTable, TxConstants.ConflictDetection conflictLevel) {
    this(hTable, conflictLevel, false);
  }

  /**
   * Create a transactional aware instance of the passed HTable, with the option of allowing
   * non-transactional operations.
   * @param hTable underlying HBase table to use
   * @param allowNonTransactional if true, additional operations (checkAndPut, increment,
   *          checkAndDelete) will be available, though non-transactional
   */
  public TransactionAwareHTable(Table hTable, boolean allowNonTransactional) {
    this(hTable, TxConstants.ConflictDetection.COLUMN, allowNonTransactional);
  }

  /**
   * Create a transactional aware instance of the passed HTable, with the option of allowing
   * non-transactional operations.
   * @param hTable underlying HBase table to use
   * @param conflictLevel level of conflict detection to perform (defaults to {@code COLUMN})
   * @param allowNonTransactional if true, additional operations (checkAndPut, increment,
   *          checkAndDelete) will be available, though non-transactional
   */
  public TransactionAwareHTable(Table hTable, TxConstants.ConflictDetection conflictLevel,
      boolean allowNonTransactional) {
    super(conflictLevel, allowNonTransactional,
        hTable.getConfiguration().getBoolean(
            TxConstants.TX_PRE_014_CHANGESET_KEY,
            TxConstants.DEFAULT_TX_PRE_014_CHANGESET_KEY));
    this.hTable = hTable;
  }

  /* AbstractTransactionAwareTable implementation */

  @Override
  protected byte[] getTableKey() {
    return hTable.getName().getName();
  }

  @Override
  protected boolean doCommit() throws IOException {
    return true;
  }

  @Override
  protected boolean doRollback() throws Exception {
    try {
      // pre-size arraylist of deletes
      int size = 0;
      for (Set<ActionChange> cs : changeSets.values()) {
        size += cs.size();
      }
      List<Delete> rollbackDeletes = new ArrayList<>(size);
      for (Map.Entry<Long, Set<ActionChange>> entry : changeSets.entrySet()) {
        long transactionTimestamp = entry.getKey();
        for (ActionChange change : entry.getValue()) {
          byte[] row = change.getRow();
          byte[] family = change.getFamily();
          byte[] qualifier = change.getQualifier();
          Delete rollbackDelete = new Delete(row);
          makeRollbackOperation(rollbackDelete);
          switch (conflictLevel) {
          case ROW:
          case NONE:
            // issue family delete for the tx write pointer
            rollbackDelete.addFamilyVersion(change.getFamily(), transactionTimestamp);
            break;
          case COLUMN:
            if (family != null && qualifier == null) {
              rollbackDelete.addFamilyVersion(family, transactionTimestamp);
            } else if (family != null && qualifier != null) {
              rollbackDelete.addColumn(family, qualifier, transactionTimestamp);
            }
            break;
          default:
            throw new IllegalStateException("Unknown conflict detection level: " + conflictLevel);
          }
          rollbackDeletes.add(rollbackDelete);
        }
      }
      hTable.delete(rollbackDeletes);
      return true;
    } finally {
      tx = null;
      changeSets.clear();
    }
  }

  /* HTableInterface implementation */

  @Override
  public TableName getName() {
    return hTable.getName();
  }

  @Override
  public Configuration getConfiguration() {
    return hTable.getConfiguration();
  }

  @Override
  public HTableDescriptor getTableDescriptor() throws IOException {
    return hTable.getTableDescriptor();
  }

  @Override
  public boolean exists(Get get) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    return hTable.exists(transactionalizeAction(get));
  }

  @Override
  public void batch(List<? extends Row> actions, Object[] results)
      throws IOException, InterruptedException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    hTable.batch(transactionalizeActions(actions), results);
  }

  @Override
  public <R> void batchCallback(List<? extends Row> actions, Object[] results,
      Batch.Callback<R> callback) throws IOException, InterruptedException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    hTable.batchCallback(transactionalizeActions(actions), results, callback);
  }

  @Override
  public Result get(Get get) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    return hTable.get(transactionalizeAction(get));
  }

  @Override
  public Result[] get(List<Get> gets) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    ArrayList<Get> transactionalizedGets = new ArrayList<>();
    for (Get get : gets) {
      transactionalizedGets.add(transactionalizeAction(get));
    }
    return hTable.get(transactionalizedGets);
  }

  @Override
  public ResultScanner getScanner(Scan scan) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    return hTable.getScanner(transactionalizeAction(scan));
  }

  @Override
  public ResultScanner getScanner(byte[] family) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    Scan scan = new Scan();
    scan.addFamily(family);
    return hTable.getScanner(transactionalizeAction(scan));
  }

  @Override
  public ResultScanner getScanner(byte[] family, byte[] qualifier) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    Scan scan = new Scan();
    scan.addColumn(family, qualifier);
    return hTable.getScanner(transactionalizeAction(scan));
  }

  @Override
  public void put(Put put) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    Put txPut = transactionalizeAction(put);
    hTable.put(txPut);
  }

  @Override
  public void put(List<Put> puts) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    List<Put> transactionalizedPuts = new ArrayList<>(puts.size());
    for (Put put : puts) {
      Put txPut = transactionalizeAction(put);
      transactionalizedPuts.add(txPut);
    }
    hTable.put(transactionalizedPuts);
  }

  @Override
  public void delete(Delete delete) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    hTable.put(transactionalizeAction(delete));
  }

  @Override
  public void delete(List<Delete> deletes) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    List<Put> transactionalizedDeletes = new ArrayList<>(deletes.size());
    for (Delete delete : deletes) {
      Put txDelete = transactionalizeAction(delete);
      transactionalizedDeletes.add(txDelete);
    }
    hTable.put(transactionalizedDeletes);
  }

  @Override
  public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier, byte[] value, Put put)
      throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndPut(row, family, qualifier, value, put);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier, byte[] value,
      Delete delete) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndDelete(row, family, qualifier, value, delete);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndDelete(byte[] bytes, byte[] bytes1, byte[] bytes2,
      CompareFilter.CompareOp compareOp, byte[] bytes3, Delete delete) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndDelete(bytes, bytes1, bytes2, compareOp, bytes3, delete);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndPut(byte[] bytes, byte[] bytes1, byte[] bytes2,
      CompareFilter.CompareOp compareOp, byte[] bytes3, Put put) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndPut(bytes, bytes1, bytes2, compareOp, bytes3, put);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndMutate(byte[] row, byte[] family, byte[] qualifier,
      CompareFilter.CompareOp compareOp, byte[] value, RowMutations rowMutations)
      throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndMutate(row, family, qualifier, compareOp, value, rowMutations);
    }

    throw new UnsupportedOperationException(
        "checkAndMutate operation is not supported transactionally");
  }

  @Override
  public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier, CompareOperator op,
      byte[] value, Put put) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndPut(row, family, qualifier, value, put);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }

  }

  @Override
  public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier, CompareOperator op,
      byte[] value, Delete delete) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndDelete(row, family, qualifier, op, value, delete);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean checkAndMutate(byte[] row, byte[] family, byte[] qualifier, CompareOperator op,
      byte[] value, RowMutations mutation) throws IOException {
    if (allowNonTransactional) {
      return hTable.checkAndMutate(row, family, qualifier, op, value, mutation);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public boolean[] existsAll(List<Get> gets) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    List<Get> transactionalizedGets = new ArrayList<>(gets.size());
    for (Get get : gets) {
      transactionalizedGets.add(transactionalizeAction(get));
    }
    return hTable.existsAll(transactionalizedGets);
  }

  @Override
  public void mutateRow(RowMutations rm) throws IOException {
    if (tx == null) {
      throw new IOException("Transaction not started");
    }
    RowMutations transactionalMutations = new RowMutations(rm.getRow());
    for (Mutation mutation : rm.getMutations()) {
      if (mutation instanceof Put) {
        transactionalMutations.add(transactionalizeAction((Put) mutation));
      } else if (mutation instanceof Delete) {
        transactionalMutations.add(transactionalizeAction((Delete) mutation));
      }
    }
    hTable.mutateRow(transactionalMutations);
  }

  @Override
  public Result append(Append append) throws IOException {
    if (allowNonTransactional) {
      return hTable.append(append);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public Result increment(Increment increment) throws IOException {
    if (allowNonTransactional) {
      return hTable.increment(increment);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public long incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount)
      throws IOException {
    if (allowNonTransactional) {
      return hTable.incrementColumnValue(row, family, qualifier, amount);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public long incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount,
      Durability durability) throws IOException {
    if (allowNonTransactional) {
      return hTable.incrementColumnValue(row, family, qualifier, amount, durability);
    } else {
      throw new UnsupportedOperationException("Operation is not supported transactionally");
    }
  }

  @Override
  public void close() throws IOException {
    hTable.close();
  }

  @Override
  public CoprocessorRpcChannel coprocessorService(byte[] row) {
    return hTable.coprocessorService(row);
  }

  @Override
  public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> service, byte[] startKey,
      byte[] endKey, Batch.Call<T, R> callable) throws ServiceException, Throwable {
    return hTable.coprocessorService(service, startKey, endKey, callable);
  }

  @Override
  public <T extends Service, R> void coprocessorService(Class<T> service, byte[] startKey,
      byte[] endKey, Batch.Call<T, R> callable, Batch.Callback<R> callback)
      throws ServiceException, Throwable {
    hTable.coprocessorService(service, startKey, endKey, callable, callback);
  }

  @Override
  public <R extends Message> Map<byte[], R> batchCoprocessorService(
      MethodDescriptor methodDescriptor, Message request, byte[] startKey, byte[] endKey,
      R responsePrototype) throws ServiceException, Throwable {
    return hTable.batchCoprocessorService(methodDescriptor, request, startKey, endKey,
      responsePrototype);
  }

  @Override
  public <R extends Message> void batchCoprocessorService(MethodDescriptor methodDescriptor,
      Message request, byte[] startKey, byte[] endKey, R responsePrototype, Callback<R> callback)
      throws ServiceException, Throwable {
    hTable.batchCoprocessorService(methodDescriptor, request, startKey, endKey, responsePrototype,
      callback);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -