Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1715 (2018-03-08 14:20:20)

RetryingReadRowsOperation uses more final variables
- RowMerger is final and now has a reset method
- `CallToStreamObserverAdapter` is now final, and use required no modification.


Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
index 45f3aa721..9f97f875a 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
@@ -104,12 +104,12 @@ public class RetryingReadRowsOperation extends
   Clock clock = Clock.SYSTEM;
   private final ReadRowsRequestManager requestManager;
   private final StreamObserver<FlatRow> rowObserver;
-  private RowMerger rowMerger;
+  private final RowMerger rowMerger;
   private long lastResponseMs;
 
   // The number of times we've retried after a timeout
   private int timeoutRetryCount = 0;
-  private CallToStreamObserverAdapter adapter;
+  private final CallToStreamObserverAdapter adapter;
   private StreamObserver<ReadRowsResponse> resultObserver;
   private int totalRowsProcessed = 0;
   private volatile ReadRowsRequest nextRequest;
@@ -124,6 +124,8 @@ public class RetryingReadRowsOperation extends
       Metadata originalMetadata) {
     super(retryOptions, request, retryableRpc, callOptions, retryExecutorService, originalMetadata);
     this.rowObserver = observer;
+    this.rowMerger = new RowMerger(rowObserver);
+    this.adapter = new CallToStreamObserverAdapter();
     this.requestManager = new ReadRowsRequestManager(request);
     this.nextRequest = request;
   }
@@ -146,8 +148,6 @@ public class RetryingReadRowsOperation extends
   public void run() {
     try {
       // restart the clock.
-      this.rowMerger = new RowMerger(rowObserver);
-      adapter = new CallToStreamObserverAdapter();
       synchronized (callLock) {
         super.run();
         // pre-fetch one more result, for performance reasons.
@@ -237,10 +237,7 @@ public class RetryingReadRowsOperation extends
   /** {@inheritDoc} */
   @Override
   public void setException(Exception exception) {
-    rowObserver.onError(exception);
-    // cleanup any state that was in RowMerger. There may be a partial row in progress which needs
-    // to be reset.
-    rowMerger = new RowMerger(rowObserver);
+    rowMerger.onError(exception);
     super.setException(exception);
   }
 
@@ -313,12 +310,13 @@ public class RetryingReadRowsOperation extends
 
   @Override
   protected void performRetry(long nextBackOff) {
-    buildUpdatedRequst();
+    buildUpdatedRequest();
     super.performRetry(nextBackOff);
   }
 
   @VisibleForTesting
-  ReadRowsRequest buildUpdatedRequst() {
+  ReadRowsRequest buildUpdatedRequest() {
+    this.rowMerger.clearRowInProgress();
     return nextRequest = requestManager.buildUpdatedRequest();
   }
 
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java
index 101e34f07..28a11d4e7 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java
@@ -22,6 +22,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Objects;
 import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.bigtable.v2.ReadRowsResponse;
@@ -364,9 +365,9 @@ public class RowMerger implements StreamObserver<ReadRowsResponse> {
 
   private RowMergerState state = RowMergerState.NewRow;
   private ByteString lastCompletedRowKey = null;
-  private RowInProgress rowInProgress;
-  private boolean complete;
-  private int rowCountInLastMessage = -1;
+  private RowInProgress rowInProgress = null;
+  private boolean complete = false;
+  private Integer rowCountInLastMessage = null;
 
   /**
    * <p>Constructor for RowMerger.</p>
@@ -377,6 +378,13 @@ public class RowMerger implements StreamObserver<ReadRowsResponse> {
     this.observer = observer;
   }
 
+  public void clearRowInProgress() {
+    Preconditions.checkState(!complete, "Cannot reset Rowmerger after completion");
+    state = RowMergerState.NewRow;
+    rowInProgress = null;
+    rowCountInLastMessage = null;
+  }
+
   /** {@inheritDoc} */
   @Override
   public final void onNext(ReadRowsResponse readRowsResponse) {
@@ -384,7 +392,7 @@ public class RowMerger implements StreamObserver<ReadRowsResponse> {
       onError(new IllegalStateException("Adding partialRow after completion"));
       return;
     }
-    rowCountInLastMessage = 0;
+    int rowsProcessed = 0;
     for (int i = 0; i < readRowsResponse.getChunksCount(); i++) {
       try {
         CellChunk chunk = readRowsResponse.getChunks(i);
@@ -415,30 +423,32 @@ public class RowMerger implements StreamObserver<ReadRowsResponse> {
         if (chunk.getCommitRow()) {
           observer.onNext(rowInProgress.buildRow());
           lastCompletedRowKey = rowInProgress.getRowKey();
-          rowInProgress = null;
           state = RowMergerState.NewRow;
-          rowCountInLastMessage++;
+          rowInProgress = null;
+          rowsProcessed++;
         }
       } catch (Throwable e) {
         onError(e);
         return;
       }
     }
+    this.rowCountInLastMessage = rowsProcessed;
   }
 
   /**
    * @return the number of rows processed in the previous call to {@link #onNext(ReadRowsResponse)}.
    */
-  public int getRowCountInLastMessage() {
+  public Integer getRowCountInLastMessage() {
     return rowCountInLastMessage;
   }
 
+  public ByteString getLastCompletedRowKey() {
+    return lastCompletedRowKey;
+  }
 
-  /** {@inheritDoc} */
-  @Override
-  public void onError(Throwable e) {
-    observer.onError(e);
-    complete = true;
+  @VisibleForTesting
+  boolean isInNewState() {
+    return state == RowMergerState.NewRow && rowInProgress == null;
   }
 
   /**
@@ -448,10 +458,19 @@ public class RowMerger implements StreamObserver<ReadRowsResponse> {
    */
   @Override
   public void onCompleted() {
+    complete = true;
     state.handleOnComplete(observer);
   }
 
-  public ByteString getLastCompletedRowKey() {
-    return lastCompletedRowKey;
+  /** {@inheritDoc} */
+  @Override
+  public void onError(Throwable e) {
+    complete = true;
+    observer.onError(e);
+  }
+
+  @VisibleForTesting
+  boolean isComplete() {
+    return complete;
   }
 }
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java
index 6e7e7c625..f2c3023b5 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java
@@ -66,7 +66,7 @@ import io.opencensus.trace.Tracing;
 
 /**
  * Bigtable implementation of {@link AsyncTable}.
- * 
+ *
  * @author spollapally
  */
 public class BigtableAsyncTable implements AsyncTable<ScanResultConsumer> {
@@ -103,7 +103,7 @@ public class BigtableAsyncTable implements AsyncTable<ScanResultConsumer> {
   @Override
   public CompletableFuture<Result> append(Append append) {
     ReadModifyWriteRowRequest request = hbaseAdapter.adapt(append);
-    Function<? super ReadModifyWriteRowResponse, ? extends Result> adaptRowFunction = response -> 
+    Function<? super ReadModifyWriteRowResponse, ? extends Result> adaptRowFunction = response ->
         append.isReturnResults()
             ? Adapters.ROW_ADAPTER.adaptResponse(response.getRow())
             : null;
@@ -125,56 +125,39 @@ public class BigtableAsyncTable implements AsyncTable<ScanResultConsumer> {
 
   private final class CheckAndMutateBuilderImpl implements CheckAndMutateBuilder {
 
-    private final byte[] row;
-
-    private final byte[] family;
-
-    private byte[] qualifier;
-
-    private CompareOperator op;
-
-    private byte[] value;
+    private final CheckAndMutateUtil.RequestBuilder builder;
 
     CheckAndMutateBuilderImpl(byte[] row, byte[] family) {
-      this.row = Preconditions.checkNotNull(row, "row is null");
-      this.family = Preconditions.checkNotNull(family, "family is null");
+      this.builder = new CheckAndMutateUtil.RequestBuilder(hbaseAdapter, row, family);
     }
 
     @Override
     public CheckAndMutateBuilder qualifier(byte[] qualifier) {
-      this.qualifier = Preconditions.checkNotNull(qualifier, "qualifier is null. Consider using" +
-          " an empty byte array, or just do not call this method if you want a null qualifier");
+      builder.qualifier(qualifier);
       return this;
     }
 
     @Override
     public CheckAndMutateBuilder ifNotExists() {
-      this.op = CompareOperator.EQUAL;
-      this.value = null;
+      builder.ifNotExists();
       return this;
     }
 
     @Override
     public CheckAndMutateBuilder ifMatches(CompareOperator compareOp, byte[] value) {
-      this.op = Preconditions.checkNotNull(compareOp, "compareOp is null");
-      if (compareOp != CompareOperator.EQUAL && compareOp != CompareOperator.NOT_EQUAL) {
-        this.value =
-            Preconditions.checkNotNull(value, "value is null for compareOperator: " + compareOp);
-      } else {
-        this.value = value;
+      Preconditions.checkNotNull(compareOp, "compareOp is null");
+      if (compareOp != CompareOperator.NOT_EQUAL) {
+        Preconditions.checkNotNull(value, "value is null for compareOperator: " + compareOp);
       }
+      builder.ifMatches(BigtableTable.toCompareOp(compareOp), value);
       return this;
     }
 
-    private void preCheck() {
-      Preconditions.checkNotNull(op, "condition is null. You need to specify the condition by" +
-          " calling ifNotExists/ifEquals/ifMatches before executing the request");
-    }
-
     @Override
     public CompletableFuture<Boolean> thenPut(Put put) {
       try {
-        return call(put.getRow(), hbaseAdapter.adapt(put));
+        builder.withPut(put);
+        return call();
       } catch (Exception e) {
         return FutureUtils.failedFuture(e);
       }
@@ -183,7 +166,8 @@ public class BigtableAsyncTable implements AsyncTable<ScanResultConsumer> {
     @Override
     public CompletableFuture<Boolean> thenDelete(Delete delete) {
       try {
-        return call(delete.getRow(), hbaseAdapter.adapt(delete));
+        builder.withDelete(delete);
+        return call();
       } catch (Exception e) {
         return FutureUtils.failedFuture(e);
       }
@@ -192,25 +176,16 @@ public class BigtableAsyncTable implements AsyncTable<ScanResultConsumer> {
     @Override
     public CompletableFuture<Boolean> thenMutate(RowMutations mutation) {
       try {
-        return call(mutation.getRow(), hbaseAdapter.adapt(mutation));
+        builder.withMutations(mutation);
+        return call();
       } catch (Exception e) {
         return FutureUtils.failedFuture(e);
       }
     }
 
-    private CompletableFuture<Boolean> call(byte[] actionRow, MutateRowRequest mutateRowRequest)
+    private CompletableFuture<Boolean> call()
         throws IOException {
-      preCheck();
-      CheckAndMutateRowRequest request =
-          CheckAndMutateUtil.makeConditionalMutationRequest(
-              hbaseAdapter,
-              row,
-              family,
-              qualifier,
-              BigtableTable.toCompareOp(op),
-              value,
-              actionRow,
-              mutateRowRequest.getMutationsList());
+      CheckAndMutateRowRequest request = builder.build();
       return client.checkAndMutateRowAsync(request).thenApply(
         response -> CheckAndMutateUtil.wasMutationApplied(request, response));
     }
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/AbstractBigtableTable.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/AbstractBigtableTable.java
index a47edc021..1297bf255 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/AbstractBigtableTable.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/AbstractBigtableTable.java
@@ -380,8 +380,11 @@ public abstract class AbstractBigtableTable implements Table {
       CompareFilter.CompareOp compareOp, byte[] value, Put put) throws IOException {
     LOG.trace("checkAndPut(byte[], byte[], byte[], CompareOp, value, Put)");
     CheckAndMutateRowRequest request =
-        CheckAndMutateUtil.makeConditionalMutationRequest(hbaseAdapter, row, family, qualifier,
-          compareOp, value, put.getRow(), hbaseAdapter.adapt(put).getMutationsList());
+        new CheckAndMutateUtil.RequestBuilder(hbaseAdapter, row, family)
+            .qualifier(qualifier)
+            .ifMatches(compareOp, value)
+            .withPut(put)
+            .build();
 
     return checkAndMutate(row, request, "checkAndPut");
   }
@@ -420,8 +423,11 @@ public abstract class AbstractBigtableTable implements Table {
       CompareFilter.CompareOp compareOp, byte[] value, Delete delete) throws IOException {
     LOG.trace("checkAndDelete(byte[], byte[], byte[], CompareOp, byte[], Delete)");
     CheckAndMutateRowRequest request =
-        CheckAndMutateUtil.makeConditionalMutationRequest(hbaseAdapter, row, family, qualifier,
-          compareOp, value, delete.getRow(), hbaseAdapter.adapt(delete).getMutationsList());
+        new CheckAndMutateUtil.RequestBuilder(hbaseAdapter, row, family)
+            .qualifier(qualifier)
+            .ifMatches(compareOp, value)
+            .withDelete(delete)
+            .build();
 
     return checkAndMutate(row, request, "checkAndDelete");
   }
@@ -435,8 +441,11 @@ public abstract class AbstractBigtableTable implements Table {
     LOG.trace("checkAndMutate(byte[], byte[], byte[], CompareOp, byte[], RowMutations)");
 
     CheckAndMutateRowRequest request =
-        CheckAndMutateUtil.makeConditionalMutationRequest(hbaseAdapter, row, family, qualifier,
-          compareOp, value, rm.getRow(), hbaseAdapter.adapt(rm).getMutationsList());
+        new CheckAndMutateUtil.RequestBuilder(hbaseAdapter, row, family)
+            .qualifier(qualifier)
+            .ifMatches(compareOp, value)
+            .withMutations(rm)
+            .build();
 
     return checkAndMutate(row, request, "checkAndMutate");
   }
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/CheckAndMutateUtil.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/CheckAndMutateUtil.java
index 322aa8d25..0d38d1e0a 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/CheckAndMutateUtil.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/CheckAndMutateUtil.java
@@ -15,25 +15,27 @@
  */
 package com.google.cloud.bigtable.hbase.adapters;
 
-import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
+import com.google.bigtable.v2.*;
+import com.google.common.base.Preconditions;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RowMutations;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.filter.BinaryComparator;
-import org.apache.hadoop.hbase.filter.CompareFilter;
 import org.apache.hadoop.hbase.filter.ValueFilter;
 import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
 
-import com.google.bigtable.v2.CheckAndMutateRowRequest;
-import com.google.bigtable.v2.CheckAndMutateRowResponse;
-import com.google.bigtable.v2.Mutation;
-import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.cloud.bigtable.hbase.adapters.read.ReadHooks;
 import com.google.common.base.Function;
 import com.google.protobuf.ByteString;
 
+import javax.annotation.Nullable;
+
 public class CheckAndMutateUtil {
 
   // ReadHooks don't make sense from conditional mutations. If any filter attempts to make use of
@@ -71,67 +73,147 @@ public class CheckAndMutateUtil {
         || (request.getFalseMutationsCount() > 0
         && !response.getPredicateMatched());
   }
-  
+
   /**
-   * <p>
-   * makeConditionalMutationRequest.
-   * </p>
-   * @param hbaseAdapter a {@link HBaseRequestAdapter} used to convert HBase
-   *          {@link org.apache.hadoop.hbase.client.Mutation} to Cloud Bigtable {@link Mutation}
-   * @param row an array of byte.
-   * @param family an array of byte.
-   * @param qualifier an array of byte.
-   * @param compareOp a {@link org.apache.hadoop.hbase.filter.CompareFilter.CompareOp} object.
-   * @param value an array of byte.
-   * @param actionRow an array of byte.
-   * @param mutations a {@link java.util.List} object.
-   * @return a {@link com.google.bigtable.v2.CheckAndMutateRowRequest.Builder} object.
-   * @throws java.io.IOException if any.
+   * This class can be used to convert HBase checkAnd* operations to Bigtable
+   * {@link CheckAndMutateRowRequest}s.
    */
-  public static CheckAndMutateRowRequest makeConditionalMutationRequest(
-      HBaseRequestAdapter hbaseAdapter,
-      byte[] row,
-      byte[] family,
-      byte[] qualifier,
-      CompareFilter.CompareOp compareOp,
-      byte[] value,
-      byte[] actionRow,
-      List<com.google.bigtable.v2.Mutation> mutations) throws IOException {
-    if (!Arrays.equals(actionRow, row)) {
-      // The following odd exception message is for compatibility with HBase.
-      throw new DoNotRetryIOException("Action's getRow must match the passed row");
+  public static class RequestBuilder {
+    private final HBaseRequestAdapter hbaseAdapter;
+    private final CheckAndMutateRowRequest.Builder requestBuilder;
+
+    private final List<Mutation> mutations = new ArrayList<>();
+
+    private final byte[] row;
+    private final byte[] family;
+    private byte[] qualifier;
+    private CompareOp compareOp;
+    private byte[] value;
+    private boolean checkNonExistence = false;
+
+    /**
+     * <p>
+     * RequestBuilder.
+     * </p>
+     * @param hbaseAdapter a {@link HBaseRequestAdapter} used to convert HBase
+     *          {@link org.apache.hadoop.hbase.client.Mutation} to Cloud Bigtable {@link Mutation}
+     * @param row the RowKey in which to to check value matching
+     * @param family the family in which to check value matching.
+     */
+    public RequestBuilder(HBaseRequestAdapter hbaseAdapter, byte[] row, byte[] family) {
+      this.row = Preconditions.checkNotNull(row, "row is null");
+      this.family = Preconditions.checkNotNull(family, "family is null");
+
+      requestBuilder = CheckAndMutateRowRequest.newBuilder()
+          .setRowKey(ByteString.copyFrom(row))
+          .setTableName(hbaseAdapter.getBigtableTableName().toString());
+
+      // The hbaseAdapter used here should not set client-side timestamps, since that may cause strange contention
+      // issues.  See issue #1709.
+      this.hbaseAdapter = hbaseAdapter.withServerSideTimestamps();
+    }
+
+    public RequestBuilder qualifier(byte[] qualifier) {
+      this.qualifier = Preconditions.checkNotNull(qualifier, "qualifier is null. Consider using" +
+          " an empty byte array, or just do not call this method if you want a null qualifier");
+      return this;
+    }
+
+    public RequestBuilder ifNotExists() {
+      Preconditions.checkState(compareOp == null,
+          "ifNotExists and ifMatches are mutually exclusive");
+      this.checkNonExistence = true;
+      return this;
+    }
+
+    /**
+     * For non-null values, this produces a {@link Filter} equivalent to:
+     * <pre>
+     *   new ValueFilter(reverseCompareOp(compareOp), new BinaryComparator(value)
+     * </pre>
+     * <p>
+     * Null values are bit tricky. In HBase 1.* style check and mutate, value == null always means `check non-existence`
+     * regardless of compareOp.  That's semantically confusing for CompareOperators other than EQUALS.
+     * It's even more confusing if value == null and compareOp = NOT_EQUALS.
+     * <p>
+     * HBase 2.* APIs introduced an explicit method of checking for "Non-Existence" as an independent concept from
+     * compareOp, which is the inspiration for the {@link #ifNotExists()} method.
+     * <p>
+     * Checking for existence in HBase can be expressed as follows:
+     * <pre>
+     * compareOp = CompareOp.GREATER_OR_EQUAL, value = new byte[]{Byte.MIN_VALUE})
+     * </pre>
+     * <p>
+     * Bigtable decided that value == null with a compareOp of NOT_EQUALS actually means check for existence, even
+     * though HBase will still treat it as non-existence.
+     *
+     * @param compareOp a {@link CompareOp}
+     * @param value
+     * @return this
+     */
+    public RequestBuilder ifMatches(CompareOp compareOp, @Nullable byte[] value) {
+      Preconditions.checkState(checkNonExistence == false,
+          "ifNotExists and ifMatches are mutually exclusive");
+
+      this.compareOp = Preconditions.checkNotNull(compareOp, "compareOp is null");
+
+      // TODO (issue #1704): only NOT_EQUALS should allow null.  Everything else should use ifNotExists();
+      this.value = value;
+      return this;
+    }
+
+    public RequestBuilder withPut(Put put) throws DoNotRetryIOException {
+      return addMutations(put.getRow(), hbaseAdapter.adapt(put).getMutationsList());
+    }
+
+    public RequestBuilder withDelete(Delete delete) throws DoNotRetryIOException {
+      return addMutations(delete.getRow(), hbaseAdapter.adapt(delete).getMutationsList());
+    }
+
+    public RequestBuilder withMutations(RowMutations rm) throws DoNotRetryIOException {
+      return addMutations(rm.getRow(), hbaseAdapter.adapt(rm).getMutationsList());
     }
 
-    CheckAndMutateRowRequest.Builder requestBuilder =
-        CheckAndMutateRowRequest.newBuilder();
-
-    requestBuilder.setTableName(hbaseAdapter.getBigtableTableName().toString());
-
-    requestBuilder.setRowKey(ByteString.copyFrom(row));
-    Scan scan = new Scan().addColumn(family, qualifier);
-    scan.setMaxVersions(1);
-    if (value == null) {
-      // If we don't have a value and we are doing CompareOp.EQUAL, we want to mutate if there
-      // is no cell with the qualifier. If we are doing CompareOp.NOT_EQUAL, we want to mutate
-      // if there is any cell. We don't actually want an extra filter for either of these cases,
-      // but we do need to invert the compare op.
-      if (CompareFilter.CompareOp.EQUAL.equals(compareOp)) {
-        requestBuilder.addAllFalseMutations(mutations);
-      } else if (CompareFilter.CompareOp.NOT_EQUAL.equals(compareOp)) {
+    private RequestBuilder addMutations(byte[] actionRow, List<Mutation> mutations) throws DoNotRetryIOException {
+      if (!Arrays.equals(actionRow, row)) {
+        // The following odd exception message is for compatibility with HBase.
+        throw new DoNotRetryIOException("Action's getRow must match the passed row");
+      }
+      this.mutations.addAll(mutations);
+      return this;
+    }
+
+    public CheckAndMutateRowRequest build() {
+      Preconditions.checkState(checkNonExistence || compareOp != null,
+          "condition is null. You need to specify the condition by" +
+          " calling ifNotExists/ifEquals/ifMatches before executing the request");
+
+      Scan scan = new Scan();
+      scan.setMaxVersions(1);
+      scan.addColumn(family, qualifier);
+
+      if (value == null || checkNonExistence) {
+        // See ifMatches javadoc for more information on this
+        if (CompareOp.NOT_EQUAL.equals(compareOp)) {
+          // check for existence
+          requestBuilder.addAllTrueMutations(mutations);
+        } else {
+          // check for non-existence
+          requestBuilder.addAllFalseMutations(mutations);
+        }
+      } else {
+        ValueFilter valueFilter =
+            new ValueFilter(reverseCompareOp(compareOp), new BinaryComparator(value));
+        scan.setFilter(valueFilter);
         requestBuilder.addAllTrueMutations(mutations);
       }
-    } else {
-      ValueFilter valueFilter =
-          new ValueFilter(reverseCompareOp(compareOp), new BinaryComparator(value));
-      scan.setFilter(valueFilter);
-      requestBuilder.addAllTrueMutations(mutations);
+      requestBuilder.setPredicateFilter(
+          Adapters.SCAN_ADAPTER.buildFilter(scan, UNSUPPORTED_READ_HOOKS));
+
+      return requestBuilder.build();
     }
-    requestBuilder.setPredicateFilter(
-      Adapters.SCAN_ADAPTER.buildFilter(scan, UNSUPPORTED_READ_HOOKS));
-    return requestBuilder.build();
   }
 
-
   /**
    * For some reason, the ordering of CheckAndMutate operations is the inverse order of normal
    * {@link ValueFilter} operations.
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/HBaseRequestAdapter.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/HBaseRequestAdapter.java
index 6976a892c..2faa81c98 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/HBaseRequestAdapter.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/HBaseRequestAdapter.java
@@ -15,6 +15,7 @@
  */
 package com.google.cloud.bigtable.hbase.adapters;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Append;
@@ -48,10 +49,19 @@ public class HBaseRequestAdapter {
     protected final RowMutationsAdapter rowMutationsAdapter;
 
     public MutationAdapters(BigtableOptions options, Configuration config) {
-      this.putAdapter = Adapters.createPutAdapter(config, options);
+      this(Adapters.createPutAdapter(config, options));
+    }
+
+    @VisibleForTesting
+    MutationAdapters(PutAdapter putAdapter) {
+      this.putAdapter = putAdapter;
       this.hbaseMutationAdapter = Adapters.createMutationsAdapter(putAdapter);
       this.rowMutationsAdapter = new RowMutationsAdapter(hbaseMutationAdapter);
     }
+
+    public MutationAdapters withServerSideTimestamps() {
+      return new MutationAdapters(putAdapter.withServerSideTimestamps());
+    }
   }
 
   protected final MutationAdapters mutationAdapters;
@@ -72,18 +82,39 @@ public class HBaseRequestAdapter {
   /**
    * <p>Constructor for HBaseRequestAdapter.</p>
    *
-   * @param options a {@link com.google.cloud.bigtable.config.BigtableOptions} object.
-   * @param tableName a {@link org.apache.hadoop.hbase.TableName} object.
-   * @param mutationAdapters a {@link com.google.cloud.bigtable.hbase.adapters.HBaseRequestAdapter.MutationAdapters} object.
+   * @param options a {@link BigtableOptions} object.
+   * @param tableName a {@link TableName} object.
+   * @param mutationAdapters a {@link MutationAdapters} object.
    */
-  public HBaseRequestAdapter(BigtableOptions options, TableName tableName,
-      MutationAdapters mutationAdapters) {
+  public HBaseRequestAdapter(BigtableOptions options,
+                             TableName tableName,
+                             MutationAdapters mutationAdapters) {
+    this(tableName,
+        options.getInstanceName().toTableName(tableName.getQualifierAsString()),
+        mutationAdapters);
+  }
+
+
+  /**
+   * <p>Constructor for HBaseRequestAdapter.</p>
+   *
+   * @param tableName a {@link TableName} object.
+   * @param bigtableTableName a {@link BigtableTableName} object.
+   * @param mutationAdapters a {@link MutationAdapters} object.
+   */
+  @VisibleForTesting
+  HBaseRequestAdapter(TableName tableName,
+                              BigtableTableName bigtableTableName,
+                              MutationAdapters mutationAdapters) {
     this.tableName = tableName;
-    this.bigtableTableName =
-        options.getInstanceName().toTableName(tableName.getQualifierAsString());
+    this.bigtableTableName = bigtableTableName;
     this.mutationAdapters = mutationAdapters;
   }
 
+  public HBaseRequestAdapter withServerSideTimestamps(){
+    return new HBaseRequestAdapter(tableName, bigtableTableName, mutationAdapters.withServerSideTimestamps());
+  }
+
   /**
    * <p>adapt.</p>
    *
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/PutAdapter.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/PutAdapter.java
index 15000818a..6cc2864c5 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/PutAdapter.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/PutAdapter.java
@@ -72,6 +72,10 @@ public class PutAdapter extends MutationAdapter<Put> {
     this.setClientTimestamp = setClientTimestamp;
   }
 
+  PutAdapter withServerSideTimestamps() {
+    return new PutAdapter(maxKeyValueSize, false);
+  }
+
   @Override
   protected Collection<Mutation> adaptMutations(Put operation) {
     if (operation.isEmpty()) {


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
index 95bb5a00a..ffd03941c 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
@@ -82,12 +82,12 @@ public class RetryingReadRowsOperationTest {
     Builder builder = ReadRowsResponse.newBuilder();
     for (ByteString key : keys) {
       builder.addChunks(CellChunk.newBuilder()
-        .setRowKey(key)
-        .setFamilyName(StringValue.newBuilder().setValue("Family"))
-        .setQualifier(BytesValue.newBuilder().setValue(ByteString.copyFrom("qualifier", "UTF-8")))
-        .setValue(ByteString.copyFrom("value", "UTF-8"))
-        .setCommitRow(true)
-        .build());
+          .setRowKey(key)
+          .setFamilyName(StringValue.newBuilder().setValue("Family"))
+          .setQualifier(BytesValue.newBuilder().setValue(ByteString.copyFrom("qualifier", "UTF-8")))
+          .setValue(ByteString.copyFrom("value", "UTF-8"))
+          .setCommitRow(true)
+          .build());
     }
     return builder.build();
   }
@@ -231,9 +231,8 @@ public class RetryingReadRowsOperationTest {
     ByteString key1 = ByteString.copyFrom("SomeKey1", "UTF-8");
     ByteString key2 = ByteString.copyFrom("SomeKey2", "UTF-8");
     underTest.onMessage(buildResponse(key1));
-    RowMerger rw1 = underTest.getRowMerger();
     underTest.onClose(Status.ABORTED, new Metadata());
-    Assert.assertNotSame(rw1, underTest.getRowMerger());
+    Assert.assertFalse(underTest.getRowMerger().isComplete());
     underTest.onMessage(buildResponse(key2));
     verify(mockFlatRowObserver, times(2)).onNext(any(FlatRow.class));
     checkRetryRequest(underTest, key2, 8);
@@ -248,16 +247,23 @@ public class RetryingReadRowsOperationTest {
     start(underTest);
 
     final AtomicLong time = setupClock(underTest);
+    ByteString key0 = ByteString.copyFrom("SomeKey0", "UTF-8");
     ByteString key1 = ByteString.copyFrom("SomeKey1", "UTF-8");
     ByteString key2 = ByteString.copyFrom("SomeKey2", "UTF-8");
-    underTest.onMessage(buildResponse(key1));
+
+    underTest.onMessage(buildResponse(key0));
+
+    // A partial row is found
+    underTest.onMessage(setCommitToFalse(buildResponse(key1)));
+    Assert.assertFalse(underTest.getRowMerger().isInNewState());
 
     // a round of successful retries.
-    RowMerger rw1 = underTest.getRowMerger();
     performSuccessfulScanTimeouts(underTest, time);
-    Assert.assertNotSame(rw1, underTest.getRowMerger());
+    Assert.assertTrue(underTest.getRowMerger().isInNewState());
+
     underTest.onClose(Status.ABORTED, new Metadata());
-    checkRetryRequest(underTest, key1, 9);
+    Assert.assertFalse(underTest.getRowMerger().isComplete());
+    checkRetryRequest(underTest, key0, 9);
 
     // a message gets through
     underTest.onMessage(buildResponse(key2));
@@ -268,13 +274,23 @@ public class RetryingReadRowsOperationTest {
     performSuccessfulScanTimeouts(underTest, time);
 
     checkRetryRequest(underTest, key2, 8);
-    verify(mockClientCall, times(RETRY_OPTIONS.getMaxScanTimeoutRetries() * 2 + 4)).request(eq(1));
+    verify(mockClientCall, times(RETRY_OPTIONS.getMaxScanTimeoutRetries() * 2 + 5))
+        .request(eq(1));
 
     // a succsesful finish.  There were 2 x RETRY_OPTIONS.getMaxScanTimeoutRetries() timeouts,
     // and 1 ABORTED status.
     finishOK(underTest, RETRY_OPTIONS.getMaxScanTimeoutRetries() * 2 + 1);
   }
 
+  private ReadRowsResponse setCommitToFalse(ReadRowsResponse message1) {
+    int lastIndex = message1.getChunksCount() - 1;
+    CellChunk lastChunk = message1.getChunks(lastIndex);
+    message1 = message1.toBuilder()
+        .setChunks(lastIndex, lastChunk.toBuilder().setCommitRow(false).build())
+        .build();
+    return message1;
+  }
+
   @Test
   public void testScanTimeoutFail() throws UnsupportedEncodingException, BigtableRetriesExhaustedException {
     RetryingReadRowsOperation underTest = createOperation(mockFlatRowObserver);
@@ -399,11 +415,12 @@ public class RetryingReadRowsOperationTest {
     verify(mockOperationTimerContext, times(1)).close();
     verify(mockRpcMetrics, times(expectedRetryCount)).markRetry();
     verify(mockRpcTimerContext, times(expectedRetryCount + 1)).close();
+    Assert.assertTrue(underTest.getRowMerger().isComplete());
   }
 
   private static void checkRetryRequest(RetryingReadRowsOperation underTest, ByteString key,
       int rowCount) {
-    ReadRowsRequest request = underTest.buildUpdatedRequst();
+    ReadRowsRequest request = underTest.buildUpdatedRequest();
     Assert.assertEquals(key, request.getRows().getRowRanges(0).getStartKeyOpen());
     Assert.assertEquals(rowCount, request.getRowsLimit());
   }
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncCheckAndMutate.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncCheckAndMutate.java
index c92d7c18d..cd9f7e575 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncCheckAndMutate.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncCheckAndMutate.java
@@ -98,12 +98,8 @@ public class TestAsyncCheckAndMutate extends AbstractTestCheckAndMutate {
       CompareOp op, byte[] value) throws InterruptedException, ExecutionException {
     CheckAndMutateBuilder builder = getDefaultAsyncTable().checkAndMutate(row,family)
         .qualifier(qualifier);
-    if (op == CompareOp.EQUAL) {
-      if (value == null) {
-        return builder.ifNotExists();
-      } else {
-        return builder.ifEquals(value);
-      }
+    if (value == null && op != CompareOp.NOT_EQUAL) {
+      return builder.ifNotExists();
     } else {
       return builder.ifMatches(TestCheckAndMutateHBase2.translate(op), value);
     }
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-common/src/main/java/com/google/cloud/bigtable/hbase/AbstractTestCheckAndMutate.java b/bigtable-hbase-parent/bigtable-hbase-integration-tests-common/src/main/java/com/google/cloud/bigtable/hbase/AbstractTestCheckAndMutate.java
index 20416af26..9106b428b 100644
--- a/bigtable-hbase-parent/bigtable-hbase-integration-tests-common/src/main/java/com/google/cloud/bigtable/hbase/AbstractTestCheckAndMutate.java
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-common/src/main/java/com/google/cloud/bigtable/hbase/AbstractTestCheckAndMutate.java
@@ -358,12 +358,86 @@ public abstract class AbstractTestCheckAndMutate extends AbstractTest {
       CompareOp.NOT_EQUAL, Bytes.toBytes(4000l), someRandomPut);
     Assert.assertTrue("4000 != 2000 should succeed", success);
 
-    if (sharedTestEnv.isBigtable()) {
-      // This doesn't work in HBase, but we need this to work in CBT
-      success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, qualToCheck,
-        CompareOp.NOT_EQUAL, null, someRandomPut);
-      Assert.assertTrue("4000 != null should succeed", success);
+    success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, qualToCheck,
+        CompareOp.GREATER_OR_EQUAL, new byte[]{Byte.MIN_VALUE}, someRandomPut);
+    Assert.assertTrue("4000 > MIN_BYTES should succeed", success);
+  }
+
+  @Test
+  public void testCompareOpsNull() throws Exception {
+    byte[] rowKey = dataHelper.randomData("rowKey-");
+    byte[] nullQual = dataHelper.randomData("null-");
+    byte[] popluatedQual = dataHelper.randomData("pupulated-");
+    byte[] otherQual = dataHelper.randomData("other-");
+    boolean success;
+
+    Table table = getDefaultTable();
+
+    table.put(new Put(rowKey).addColumn(SharedTestEnvRule.COLUMN_FAMILY, popluatedQual,
+        Bytes.toBytes(2000l)));
+
+    Put someRandomPut =
+        new Put(rowKey).addColumn(SharedTestEnvRule.COLUMN_FAMILY, otherQual, Bytes.toBytes(1l));
+
+    success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, nullQual,
+        CompareOp.LESS, null, someRandomPut);
+    Assert.assertTrue("< null should succeed", success);
+
+    success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, nullQual,
+        CompareOp.GREATER, null, someRandomPut);
+    Assert.assertTrue("> null should succeed", success);
+
+    success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, nullQual,
+        CompareOp.LESS_OR_EQUAL, null, someRandomPut);
+    Assert.assertTrue("<= null should succeed", success);
+
+    success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, nullQual,
+        CompareOp.GREATER_OR_EQUAL, null, someRandomPut);
+    Assert.assertTrue(">= null should succeed", success);
+
+    success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, nullQual,
+        CompareOp.EQUAL, null, someRandomPut);
+    Assert.assertTrue("== null should succeed", success);
+
+    success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, nullQual,
+        CompareOp.GREATER_OR_EQUAL, new byte[]{Byte.MIN_VALUE}, someRandomPut);
+    Assert.assertFalse("> MIN_BYTES should fail", success);
+  }
+
+  /**
+   * Bigtable supports a notion of `NOT_EQUALS null` checks equating with `check for existence`.  HBase
+   * does not support this notion, and always uses `{any comparator} null` to mean `check for non-existence`.
+   */
+  @Test
+  public void testNotEqualsNull_BigtableOnly() throws Exception {
+    if (!sharedTestEnv.isBigtable()) {
+      return;
     }
+
+    byte[] rowKey = dataHelper.randomData("rowKey-");
+    byte[] nullQual = dataHelper.randomData("null-");
+    byte[] popluatedQual = dataHelper.randomData("pupulated-");
+    byte[] otherQual = dataHelper.randomData("other-");
+
+    Table table = getDefaultTable();
+
+    table.put(new Put(rowKey).addColumn(SharedTestEnvRule.COLUMN_FAMILY, popluatedQual,
+        Bytes.toBytes(2000l)));
+
+    Put someRandomPut =
+        new Put(rowKey).addColumn(SharedTestEnvRule.COLUMN_FAMILY, otherQual, Bytes.toBytes(1l));
+
+    boolean success;
+
+    // This doesn't work in HBase, but we need this to work in CBT
+    success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, nullQual,
+        CompareOp.NOT_EQUAL, null, someRandomPut);
+    Assert.assertFalse("!= null should fail", success);
+
+    // This doesn't work in HBase, but we need this to work in CBT
+    success = checkAndPut(rowKey, SharedTestEnvRule.COLUMN_FAMILY, popluatedQual,
+        CompareOp.NOT_EQUAL, null, someRandomPut);
+    Assert.assertTrue("2000 != null should succeed", success);
   }
 
   @Test
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/TestCheckAndMutateUtil.java b/bigtable-hbase-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/TestCheckAndMutateUtil.java
new file mode 100644
index 000000000..6b050c9c9
--- /dev/null
+++ b/bigtable-hbase-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/TestCheckAndMutateUtil.java
@@ -0,0 +1,298 @@
+/*
+ * Copyright 2018 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.hbase.adapters;
+
+import static com.google.cloud.bigtable.data.v2.wrappers.Filters.FILTERS;
+import com.google.bigtable.v2.CheckAndMutateRowRequest;
+import com.google.bigtable.v2.Mutation;
+import com.google.bigtable.v2.RowFilter;
+import com.google.cloud.bigtable.data.v2.wrappers.Filters;
+import com.google.cloud.bigtable.grpc.BigtableInstanceName;
+import com.google.cloud.bigtable.grpc.BigtableTableName;
+import com.google.protobuf.ByteString;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RowMutations;
+import org.apache.hadoop.hbase.filter.CompareFilter;
+import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+@RunWith(JUnit4.class)
+public class TestCheckAndMutateUtil {
+
+  private static final TableName TABLE_NAME = TableName.valueOf("SomeTable");
+  private static final BigtableTableName BT_TABLE_NAME =
+      new BigtableInstanceName("project", "instance")
+          .toTableName(TABLE_NAME.getNameAsString());
+
+  private static final byte[] rowKey = Bytes.toBytes("rowKey");
+  private static final byte[] family = Bytes.toBytes("family");
+  private static final byte[] qual = Bytes.toBytes("qual");
+  private static final byte[] checkValue = Bytes.toBytes(4);
+  private static final byte[] newValue = Bytes.toBytes(5);
+  private static final Put PUT = new Put(rowKey).addColumn(family, qual, newValue);
+  private static final Filters.ChainFilter FAMILY_AND_QUAL_FILTER = FILTERS.chain()
+      .filter(FILTERS.family().regex("family"))
+      .filter(FILTERS.qualifier().regex("qual"));
+
+  private static HBaseRequestAdapter requestAdapter;
+
+  @BeforeClass
+  public static void setup() {
+    PutAdapter putAdapter = new PutAdapter(100, true);
+    HBaseRequestAdapter.MutationAdapters mutationAdapters =
+        new HBaseRequestAdapter.MutationAdapters(putAdapter);
+    requestAdapter = new HBaseRequestAdapter(TABLE_NAME, BT_TABLE_NAME, mutationAdapters);
+  }
+
+  private static void checkPredicate(CheckAndMutateRowRequest result) {
+    RowFilter expected = FILTERS.chain()
+        .filter(FAMILY_AND_QUAL_FILTER)
+        .filter(FILTERS.limit().cellsPerColumn(1))
+        .filter(FILTERS.value().range()
+            .startClosed(ByteString.copyFrom(checkValue))
+            .endClosed(ByteString.copyFrom(checkValue))
+        )
+        .toProto();
+    Assert.assertEquals(expected, result.getPredicateFilter());
+  }
+
+  private static void checkPutMutation(Mutation mutation) {
+    Mutation.SetCell setCell = mutation.getSetCell();
+    Assert.assertArrayEquals(family, setCell.getFamilyNameBytes().toByteArray());
+    Assert.assertArrayEquals(qual, setCell.getColumnQualifier().toByteArray());
+    Assert.assertArrayEquals(newValue, setCell.getValue().toByteArray());
+  }
+
+
+  private static CheckAndMutateUtil.RequestBuilder createRequestBuilder() {
+    return new CheckAndMutateUtil.RequestBuilder(requestAdapter, rowKey, family);
+  }
+
+  //  ************************** TEST METHODS ***************************
+
+  @Test
+  /** Tests that a CheckAndMutate with a {@link Put} works correctly for the filter and mutation aspects */
+  public void testPut() throws DoNotRetryIOException {
+    CheckAndMutateUtil.RequestBuilder underTest = createRequestBuilder();
+
+    CheckAndMutateRowRequest result = underTest
+        .qualifier(qual)
+        .ifMatches(CompareOp.EQUAL, checkValue)
+        .withPut(PUT)
+        .build();
+
+    Assert.assertEquals(1, result.getTrueMutationsCount());
+
+    checkPutMutation(result.getTrueMutations(0));
+    checkPredicate(result);
+  }
+
+  @Test
+  /** Tests that a CheckAndMutate with a {@link Delete} works correctly for the filter and mutation aspects */
+  public void testDelete() throws DoNotRetryIOException {
+    CheckAndMutateUtil.RequestBuilder underTest = createRequestBuilder();
+
+    CheckAndMutateRowRequest result = underTest
+        .qualifier(qual)
+        .ifMatches(CompareOp.EQUAL, checkValue)
+        .withDelete(new Delete(rowKey).addColumns(family, qual))
+        .build();
+
+    Assert.assertEquals(1, result.getTrueMutationsCount());
+
+    Mutation.DeleteFromColumn delete = result.getTrueMutations(0).getDeleteFromColumn();
+    Assert.assertArrayEquals(family, delete.getFamilyNameBytes().toByteArray());
+    Assert.assertArrayEquals(qual, delete.getColumnQualifier().toByteArray());
+
+    checkPredicate(result);
+  }
+
+  @Test
+  /** Tests that a CheckAndMutate with a {@link RowMutations} works correctly for the filter and mutation aspects */
+  public void testRowMutations() throws IOException {
+    CheckAndMutateUtil.RequestBuilder underTest = createRequestBuilder();
+
+    RowMutations rowMutations = new RowMutations(rowKey);
+    rowMutations.add(PUT);
+    CheckAndMutateRowRequest result = underTest
+        .qualifier(qual)
+        .ifMatches(CompareOp.EQUAL, checkValue)
+        .withMutations(rowMutations)
+        .build();
+
+    Assert.assertEquals(1, result.getTrueMutationsCount());
+    checkPutMutation(result.getTrueMutations(0));
+    checkPredicate(result);
+  }
+
+  @Test
+  /**
+   * Tests that a CheckAndMutate with a {@link Put} which ensures that the conversion to a
+   * {@link CheckAndMutateRowRequest} sets a server-side timeatamp (-1) on the {@link Mutation}
+   */
+  public void testPutServerSideTimestamps() throws DoNotRetryIOException {
+    CheckAndMutateUtil.RequestBuilder underTest = createRequestBuilder();
+
+    CheckAndMutateRowRequest result = underTest
+        .qualifier(qual)
+        .ifMatches(CompareOp.EQUAL, checkValue)
+        .withPut(PUT)
+        .build();
+
+    Assert.assertEquals(1, result.getTrueMutationsCount());
+
+    Mutation.SetCell setCell = result.getTrueMutations(0).getSetCell();
+    Assert.assertEquals(-1, setCell.getTimestampMicros());
+  }
+
+  @Test
+  /**
+   * Tests that a CheckAndMutate with a {@link Put} which ensures that the conversion to a
+   * {@link CheckAndMutateRowRequest} sets a clientr-side timeatamp on the {@link Mutation}
+   * if a user explicitly sets a timestamp on the Put
+   */
+  public void testPutServerClientTimestamps() throws DoNotRetryIOException {
+    CheckAndMutateUtil.RequestBuilder underTest = createRequestBuilder();
+
+    long timestamp = (long) (Math.random() * 10000000000L);
+    CheckAndMutateRowRequest result = underTest
+        .qualifier(qual)
+        .ifMatches(CompareOp.EQUAL, checkValue)
+        .withPut(new Put(rowKey).addColumn(family, qual, timestamp, newValue))
+        .build();
+
+    Assert.assertEquals(1, result.getTrueMutationsCount());
+
+    Mutation.SetCell setCell = result.getTrueMutations(0).getSetCell();
+    Assert.assertEquals(TimeUnit.MILLISECONDS.toMicros(timestamp), setCell.getTimestampMicros());
+  }
+
+  @Test
+  /**
+   * Tests that a CheckAndMutate with a {@link RowMutations} with a {@link Put} which ensures that
+   * the conversion to a {@link CheckAndMutateRowRequest} sets a server-side timeatamp (-1) on the
+   * {@link Mutation}.
+   */
+  public void testRowMutationServerSideTimestamps() throws IOException {
+    CheckAndMutateUtil.RequestBuilder underTest = createRequestBuilder();
+
+    RowMutations rowMutations = new RowMutations(rowKey);
+    rowMutations.add(PUT);
+    CheckAndMutateRowRequest result = underTest
+        .qualifier(qual)
+        .ifMatches(CompareOp.EQUAL, checkValue)
+        .withMutations(rowMutations)
+        .build();
+
+    Assert.assertEquals(1, result.getTrueMutationsCount());
+
+    Mutation.SetCell setCell = result.getTrueMutations(0).getSetCell();
+    Assert.assertEquals(-1, setCell.getTimestampMicros());
+  }
+
+  @Test
+  /**
+   * Test to make sure that {@link CheckAndMutateUtil.RequestBuilder#ifNotExists()} works correctly
+   */
+  public void testIfNotExists() throws DoNotRetryIOException {
+    CheckAndMutateUtil.RequestBuilder underTest = createRequestBuilder();
+
+    CheckAndMutateRowRequest result = underTest
+        .qualifier(qual)
+        .ifNotExists()
+        .withPut(PUT)
+        .build();
+
+    Assert.assertEquals(1, result.getFalseMutationsCount());
+    checkPutMutation(result.getFalseMutations(0));
+
+    RowFilter expected = FILTERS.chain()
+        .filter(FAMILY_AND_QUAL_FILTER)
+        .filter(FILTERS.limit().cellsPerColumn(1))
+        .toProto();
+    Assert.assertEquals(expected, result.getPredicateFilter());
+  }
+
+  @Test
+  /**
+   * Test to make sure that {@link CheckAndMutateUtil.RequestBuilder#ifMatches(CompareOp, byte[])}
+   * wtih {@link CompareOp#NOT_EQUAL} and a null value works as
+   * expected.
+   */
+  public void testNotEqualsNull() throws DoNotRetryIOException {
+    CheckAndMutateUtil.RequestBuilder underTest = createRequestBuilder();
+
+    CheckAndMutateRowRequest result = underTest
+        .qualifier(qual)
+        .ifMatches(CompareOp.NOT_EQUAL, null)
+        .withPut(PUT)
+        .build();
+
+    Assert.assertEquals(1, result.getTrueMutationsCount());
+
+    RowFilter expected = FILTERS.chain()
+        .filter(FAMILY_AND_QUAL_FILTER)
+        .filter(FILTERS.limit().cellsPerColumn(1))
+        .toProto();
+
+    checkPutMutation(result.getTrueMutations(0));
+    Assert.assertEquals(expected, result.getPredicateFilter());
+  }
+
+  @Test
+  /**
+   * Test to make sure that {@link CheckAndMutateUtil.RequestBuilder#ifMatches(CompareOp, byte[])}
+   * wtih {@link CompareOp#NOT_EQUAL} and a null value works as
+   * expected.
+   */
+  public void testCompareOpsOtherThanNotEqualsNull() throws DoNotRetryIOException {
+    CheckAndMutateUtil.RequestBuilder underTest = createRequestBuilder();
+
+    List<CompareOp> otherOps = new ArrayList<>(Arrays.asList(CompareOp.values()));
+
+    otherOps.remove(CompareOp.NOT_EQUAL);
+    int index = (int) (Math.random() * otherOps.size());
+    CheckAndMutateRowRequest result = underTest
+        .qualifier(qual)
+        .ifMatches(otherOps.get(index), null)
+        .withPut(PUT)
+        .build();
+
+    Assert.assertEquals(1, result.getFalseMutationsCount());
+
+    RowFilter expected = FILTERS.chain()
+        .filter(FAMILY_AND_QUAL_FILTER)
+        .filter(FILTERS.limit().cellsPerColumn(1))
+        .toProto();
+
+    checkPutMutation(result.getFalseMutations(0));
+    Assert.assertEquals(expected, result.getPredicateFilter());
+  }
+}