public void processCommand(CommandArgs commandArgs) throws Exception {
        if ( commandArgs.getOperation() != Operation.REPLAY ) {
            previousOperation = commandArgs.getOperation();
        }
        
        ExecutorService exec = Executors.newFixedThreadPool(commandArgs.threads);
        log.info("Starting command run at {}", new Date());
        long currentTime = System.currentTimeMillis();
        for (int execCount = 0; execCount < commandArgs.getExecutionCount(); execCount++) {
            doneSignal = new CountDownLatch(commandArgs.threads);
            for (int i = 0; i < commandArgs.threads; i++) {
                log.info("submitting task {}", i+1);
                exec.submit(getCommandInstance(i*commandArgs.getKeysPerThread(), commandArgs, this));
            }
            log.info("all tasks submitted for execution for execution {} of {}", execCount+1, commandArgs.getExecutionCount());
            doneSignal.await();    
        }
        log.info("Finished command run at {} total duration: {} seconds", new Date(), (System.currentTimeMillis()-currentTime)/1000);
        
        
        exec.shutdown();
        
        for (CassandraHost host : latencies.keySet()) {
            LatencyTracker latency = latencies.get(host);
            log.info("Latency for host {}:\n Op Count {} \nRecentLatencyHistogram {} \nRecent Latency Micros {} \nTotalLatencyHistogram {} \nTotalLatencyMicros {}", 
                    new Object[]{host.getName(), latency.getOpCount(), latency.getRecentLatencyHistogramMicros(), latency.getRecentLatencyMicros(),
                    latency.getTotalLatencyHistogramMicros(), latency.getTotalLatencyMicros()});
        }        
    }