public void redistributeOperation(Operation op) {
    if (op.isCancelled() || op.isTimedOut()) {
      return;
    }

    if (op.getCloneCount() >= MAX_CLONE_COUNT) {
      getLogger().warn("Cancelling operation " + op + "because it has been "
        + "retried (cloned) more than " + MAX_CLONE_COUNT + "times.");
      op.cancel();
      return;
    }

    // The operation gets redistributed but has never been actually written,
    // it we just straight re-add it without cloning.
    if (op.getState() == OperationState.WRITE_QUEUED && op.getHandlingNode() != null) {
      addOperation(op.getHandlingNode(), op);
      return;
    }

    if (op instanceof MultiGetOperationImpl) {
      for (String key : ((MultiGetOperationImpl) op).getRetryKeys()) {
        addOperation(key, opFact.get(key,
          (GetOperation.Callback) op.getCallback()));
      }
    } else if (op instanceof KeyedOperation) {
      KeyedOperation ko = (KeyedOperation) op;
      int added = 0;
      for (Operation newop : opFact.clone(ko)) {
        if (newop instanceof KeyedOperation) {
          KeyedOperation newKeyedOp = (KeyedOperation) newop;
          for (String k : newKeyedOp.getKeys()) {
            addOperation(k, newop);
            op.addClone(newop);
            newop.setCloneCount(op.getCloneCount()+1);
          }
        } else {
          newop.cancel();
          getLogger().warn("Could not redistribute cloned non-keyed " +
            "operation", newop);
        }
        added++;
      }
      assert added > 0 : "Didn't add any new operations when redistributing";
    } else {
      op.cancel();
    }
  }
Redistribute the given operation to (potentially) other nodes.

Note that operations can only be redistributed if they have not been
cancelled already, timed out already or do not have definite targets
(a key).

@param op the operation to redistribute.
src/main/java/net/spy/memcached/MemcachedConnection.java