private int persist(final Map<String, Map<StaticBuffer, KCVMutation>> subMutations) {
    BackendOperation.execute(new Callable<Boolean>() {
        @Override
        public Boolean call() throws Exception {
            log.error("PERF: === CacheTransaction.persist callable mutateMany");
            manager.mutateMany(subMutations, tx);
            log.error("PERF: === CacheTransaction.persist callable mutateMany done");
            return true;
        }

        @Override
        public String toString() {
            return "CacheMutation";
        }
    }, maxWriteTime);
    log.error("PERF: === CacheTransaction.persist submutations clear");
    subMutations.clear();
    log.error("PERF: === CacheTransaction.persist submutations clear done");
    return 0;
}

public static final<V> V executeDirect(Callable<V> exe, Duration totalWaitTime) throws BackendException {
    Preconditions.checkArgument(!totalWaitTime.isZero(),"Need to specify a positive waitTime: %s",totalWaitTime);
    long maxTime = System.currentTimeMillis()+totalWaitTime.toMillis();
    log.error("PERF: === BackendOperation.executeDirect totalWaitTime={} ms, maxTime={}", totalWaitTime.toMillis(),maxTime);
    Duration waitTime = pertubateTime(BASE_REATTEMPT_TIME);
    BackendException lastException;
    log.error("PERF: === BackendOperation.executeDirect pre while");
    while (true) {
        try {
            log.error("PERF: === BackendOperation.executeDirect exe call");
            return exe.call();
        } catch (final Throwable e) {
            log.error("PERF: === BackendOperation.executeDirect exe call threw exception {}", e.getMessage());
            //Find inner-most StorageException
            Throwable ex = e;
            BackendException storeEx = null;
            do {
                if (ex instanceof BackendException) storeEx = (BackendException)ex;
            } while ((ex=ex.getCause())!=null);
            if (storeEx!=null && storeEx instanceof TemporaryBackendException) {
                lastException = storeEx;
            } else if (e instanceof BackendException) {
                throw (BackendException)e;
            } else {
                throw new PermanentBackendException("Permanent exception while executing backend operation "+exe.toString(),e);
            }
        }
        //Wait and retry
        log.error("PERF: === BackendOperation.executeDirect wait and retry");
        assert lastException!=null;
        if (System.currentTimeMillis()+waitTime.toMillis()<maxTime) {
            log.info("Temporary exception during backend operation ["+exe.toString()+"]. Attempting backoff retry.",lastException);
            try {
                Thread.sleep(waitTime.toMillis());
            } catch (InterruptedException r) {
                log.error("PERF: === BackendOperation.executeDirect sleep interrupted");
                // added thread interrupt signal to support traversal interruption
                Thread.currentThread().interrupt();
                throw new PermanentBackendException("Interrupted while waiting to retry failed backend operation", r);
            }
        } else {
            log.error("PERF: === BackendOperation.executeDirect reached maxTime");
            break;
        }
        waitTime = pertubateTime(waitTime.multipliedBy(2));
        log.error("PERF: === BackendOperation.executeDirect wait time now {}",waitTime);
    }
    throw new TemporaryBackendException("Could not successfully complete backend operation due to repeated temporary exceptions after "+totalWaitTime,lastException);
}

public void mutateMany(Map<String, Map<StaticBuffer, KCVMutation>> mutations, StoreTransaction txh) throws BackendException {
    logger.error("HBaseStoreManager: mutateMany, begin");
    final MaskedTimestamp commitTime = new MaskedTimestamp(txh);
    // In case of an addition and deletion with identical timestamps, the
    // deletion tombstone wins.
    // http://hbase.apache.org/book/versions.html#d244e4250
    logger.error("HBaseStoreManager: mutateMany, convertToCommands");
    final Map<StaticBuffer, Pair<List<Put>, Delete>> commandsPerKey =
            convertToCommands(
                    mutations,
                    commitTime.getAdditionTime(times),
                    commitTime.getDeletionTime(times));

    final List<Row> batch = new ArrayList<>(commandsPerKey.size()); // actual batch operation
    logger.error("HBaseStoreManager: mutateMany, construct batch");
    // convert sorted commands into representation required for 'batch' operation
    for (Pair<List<Put>, Delete> commands : commandsPerKey.values()) {
        if (commands.getFirst() != null && !commands.getFirst().isEmpty())
            batch.addAll(commands.getFirst());

        if (commands.getSecond() != null)
            batch.add(commands.getSecond());
    }
    logger.error("HBaseStoreManager: mutateMany, batch additions done");
    try {
        TableMask table = null;

        try {
            logger.error("HBaseStoreManager: mutateMany, get table");
            table = cnx.getTable(tableName);
            logger.error("HBaseStoreManager: mutateMany, batch (size={})",batch.size());
            table.batch(batch, new Object[batch.size()]);
        } finally {
            IOUtils.closeQuietly(table);
        }
    } catch (IOException e) {
        throw new TemporaryBackendException(e);
    } catch (InterruptedException e) {
        throw new TemporaryBackendException(e);
    }

    logger.error("HBaseStoreManager: sleepAfterWrite, commitTime={}",commitTime);
    sleepAfterWrite(txh, commitTime);
    logger.error("HBaseStoreManager: sleepAfterWrite, done");
}