package org.apache.cassandra.db;

import org.apache.cassandra.net.ICompletable;
import org.apache.cassandra.net.Message;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;

public class RowMutationCompletion implements ICompletable
{
    private final Message message;
    private final String id;
    private final RowMutation rm;
    
    private long startTime;

    private static Logger logger = LoggerFactory.getLogger(RowMutationCompletion.class);
    
    public RowMutationCompletion(Message message, String id, RowMutation rm)
    {
    	this.startTime = System.currentTimeMillis();
    	
        this.message = message;
        this.id = id;
        this.rm = rm;
    }

    // Complete the blocked RowMutation
    @Override
    public void complete()
    {
    	long endTime = System.currentTimeMillis();
    	long deltaTime = endTime - this.startTime;
    	logger.debug("~~~~ [DEEGAN] Completion Complete (" + deltaTime + "ms )" );
        RowMutationVerbHandler.instance().applyAndRespond(message, id, rm);
    }

}

@@ -365,16 +365,103 @@
 tTime;%0A%0A
+    private int maxMutationDelay = 0; //default to 0 if the environment var is not set%0A
     priv
@@ -757,24 +757,275 @@
 is.rm = rm;%0A
+        %0A        String value = System.getenv(%22max_mutation_delay_ms%22);%0A        try%7B%0A        %09this.maxMutationDelay = Integer.parseInt(value);%0A        %7D%0A        catch(NumberFormatException e)%7B%0A        %09logger.error(e.getLocalizedMessage());%0A        %7D%0A
     %7D%0A%0A    /
@@ -1282,16 +1282,141 @@
 s )%22 );%0A
+    %09try %7B%0A    %09%09Thread.sleep((long)(Math.random()*this.maxMutationDelay));%0A    %09%7Dcatch (Exception interupt) %7B%0A    %09%09%0A    %09%7D%0A
         

add delay to mutation based on max_delay env var