private Delete createDeleteMutation(List<Cell> deleteRow, DeleteType deleteType, Long timestamp) {
    long ts;
    if (timestamp == null) {
      ts = HConstants.LATEST_TIMESTAMP;
    } else {
      ts = timestamp;
    }
    // We just need the rowkey. Get it from 1st KV.
    byte[] row = CellUtil.cloneRow(deleteRow.get(0));
    Delete delete = new Delete(row, ts);
    if (deleteType == DeleteType.FAMILY) {
      Set<byte[]> families = new TreeSet<>(Bytes.BYTES_COMPARATOR);
      for (Cell kv : deleteRow) {
        if (families.add(CellUtil.cloneFamily(kv))) {
          delete.addFamily(CellUtil.cloneFamily(kv), ts);
        }
      }
    } else if (deleteType == DeleteType.COLUMN) {
      Set<Column> columns = new HashSet<>();
      for (Cell kv : deleteRow) {
        Column column = new Column(CellUtil.cloneFamily(kv), CellUtil.cloneQualifier(kv));
        if (columns.add(column)) {
          // Making deleteColumns() calls more than once for the same cf:qualifier is not correct
          // Every call to deleteColumns() will add a new KV to the familymap which will finally
          // get written to the memstore as part of delete().
          delete.addColumns(column.family, column.qualifier, ts);
        }
      }
    } else if (deleteType == DeleteType.VERSION) {
      // When some timestamp was passed to the delete() call only one version of the column (with
      // given timestamp) will be deleted. If no timestamp passed, it will delete N versions.
      // How many versions will get deleted depends on the Scan being passed. All the KVs that
      // the scan fetched will get deleted.
      int noOfVersionsToDelete = 0;
      if (timestamp == null) {
        for (Cell kv : deleteRow) {
          delete.addColumn(CellUtil.cloneFamily(kv), CellUtil.cloneQualifier(kv),
            kv.getTimestamp());
          noOfVersionsToDelete++;
        }
      } else {
        Set<Column> columns = new HashSet<>();
        for (Cell kv : deleteRow) {
          Column column = new Column(CellUtil.cloneFamily(kv), CellUtil.cloneQualifier(kv));
          // Only one version of particular column getting deleted.
          if (columns.add(column)) {
            delete.addColumn(column.family, column.qualifier, ts);
            noOfVersionsToDelete++;
          }
        }
      }
      delete.setAttribute(NO_OF_VERSIONS_TO_DELETE, Bytes.toBytes(noOfVersionsToDelete));
    }
    return delete;
  }
Defines a protocol to delete data in bulk based on a scan. The scan can be range scan or with
conditions(filters) etc.This can be used to delete rows, column family(s), column qualifier(s) or
version(s) of columns.When delete type is FAMILY or COLUMN, which all family(s) or column(s)
getting deleted will be determined by the Scan. Scan need to select all the families/qualifiers
which need to be deleted.When delete type is VERSION, Which column(s) and version(s) to be
deleted will be determined by the Scan. Scan need to select all the qualifiers and its versions
which needs to be deleted.When a timestamp is passed only one version at that timestamp will be
deleted(even if Scan fetches many versions). When timestamp passed as null, all the versions
which the Scan selects will get deleted. <br>
Example:

<pre>
<code>
Scan scan = new Scan();
// set scan properties(rowkey range, filters, timerange etc).
HTable ht = ...;
long noOfDeletedRows = 0L;
Batch.Call&lt;BulkDeleteService, BulkDeleteResponse&gt; callable =
    new Batch.Call&lt;BulkDeleteService, BulkDeleteResponse&gt;() {
  ServerRpcController controller = new ServerRpcController();
  BlockingRpcCallback&lt;BulkDeleteResponse&gt; rpcCallback =
    new BlockingRpcCallback&lt;BulkDeleteResponse&gt;();

  public BulkDeleteResponse call(BulkDeleteService service) throws IOException {
    Builder builder = BulkDeleteRequest.newBuilder();
    builder.setScan(ProtobufUtil.toScan(scan));
    builder.setDeleteType(DeleteType.VERSION);
    builder.setRowBatchSize(rowBatchSize);
    // Set optional timestamp if needed
    builder.setTimestamp(timeStamp);
    service.delete(controller, builder.build(), rpcCallback);
    return rpcCallback.get();
  }
};
Map&lt;byte[], BulkDeleteResponse&gt; result = ht.coprocessorService(BulkDeleteService.class,
 scan.getStartRow(), scan.getStopRow(), callable);
for (BulkDeleteResponse response : result.values()) {
  noOfDeletedRows += response.getRowsDeleted();
}
</code>
</pre>
hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/BulkDeleteEndpoint.java