<NME> CoordinatorRestAPITest.java
<BEF> /*
 * Copyright 2013 LinkedIn, Inc
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package voldemort.coordinator;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

import java.io.File;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import javax.mail.internet.MimeBodyPart;
import org.junit.Test;

import voldemort.ServerTestUtils;
import voldemort.cluster.Cluster;
import voldemort.server.VoldemortServer;
import voldemort.store.socket.SocketStoreFactory;
import voldemort.store.socket.clientrequest.ClientRequestExecutorPool;

public class CoordinatorRestAPITest {

    private VoldemortServer[] servers;
    private Cluster cluster;
    public static String socketUrl = "";
    private static final String STORE_NAME = "test";
    private static final String STORES_XML = "test/common/voldemort/config/single-store.xml";
import voldemort.rest.coordinator.config.CoordinatorConfig;
import voldemort.rest.coordinator.config.FileBasedStoreClientConfigService;
import voldemort.rest.coordinator.config.StoreClientConfigService;
import voldemort.server.VoldemortServer;
import voldemort.store.socket.SocketStoreFactory;
    private CoordinatorService coordinator = null;
    private final String coordinatorURL = "http://localhost:8080";

    @Before
    public void setUp() throws Exception {
        int numServers = 1;
        public void setValue(String value) {
            this.value = value;
        }
        props.setProperty("storage.configs",
                          "voldemort.store.bdb.BdbStorageConfiguration,voldemort.store.slow.SlowStorageConfiguration");

        cluster = ServerTestUtils.startVoldemortCluster(numServers,
                                                        servers,
                                                        partitionMap,
                                                        socketStoreFactory,
                                                        true, // useNio
                                                        null,
                                                        STORES_XML,
                                                        props);

        CoordinatorConfig config = new CoordinatorConfig();
        List<String> bootstrapUrls = new ArrayList<String>();
        int partitionMap[][] = { { 0, 2, 4, 6, 1, 3, 5, 7 } };
        Properties props = new Properties();
        props.setProperty("storage.configs",
                          "voldemort.store.bdb.BdbStorageConfiguration,voldemort.store.slow.SlowStorageConfiguration");
        props.setProperty("testing.slow.queueing.get.ms", "4");
        props.setProperty("testing.slow.queueing.getall.ms", "4");
        props.setProperty("testing.slow.queueing.put.ms", "4");
        props.setProperty("testing.slow.queueing.delete.ms", "4");

        ServerTestUtils.startVoldemortCluster(numServers,
                                              servers,
                                              partitionMap,
                                              socketStoreFactory,
                                              true, // useNio
                                              null,
                                              STORES_XML,
                                              props);

        // create a copy of the config file in a temp directory and work on that
        File src = new File(FAT_CLIENT_CONFIG_FILE_PATH_ORIGINAL);
        COPY_OF_FAT_CLIENT_CONFIG_FILE = new File(TestUtils.createTempDir(),
                                                  "fat-client-config" + System.currentTimeMillis()
        }
    }

    private void doPut(String key, String payload) {
        try {
            // Create the right URL and Http connection
            HttpURLConnection conn = null;
        System.out.println("\n\n************************ Starting the Coordinator *************************");

        config.setBootstrapURLs(bootstrapUrls);
        config.setFatClientConfigPath(COPY_OF_FAT_CLIENT_CONFIG_FILE.getAbsolutePath());
        StoreClientConfigService storeClientConfigs = null;
        switch(config.getFatClientConfigSource()) {
            case FILE:
                storeClientConfigs = new FileBasedStoreClientConfigService(config);
                break;
            conn.setRequestProperty("Content-Length", "" + payload.length());
            conn.setRequestProperty(VoldemortHttpRequestHandler.X_VOLD_REQUEST_TIMEOUT_MS, "1000");

            // Write the payload
            OutputStream out = conn.getOutputStream();
            out.write(payload.getBytes());
    @After
    public void tearDown() throws Exception {
        if(this.socketStoreFactory != null) {
            this.socketStoreFactory.close();
        }
                fail("Incorrect response received for a HTTP put request :"
                     + conn.getResponseCode());
            }
        } catch(Exception e) {
            e.printStackTrace();
            fail("Error in sending the REST request");
        }
    }

    private boolean doDelete(String key) {

    public static String generateRandomString(int length, ValueType type) {

        StringBuffer buffer = new StringBuffer();
        String characters = "";

        switch(type) {

            case ALPHA:
                characters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
                break;

            case ALPHANUMERIC:
                characters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890";
                break;

            case NUMERIC:
                characters = "1234567890";
                break;
        }

        int charactersLength = characters.length();

        for(int i = 0; i < length; i++) {
            double index = Math.random() * charactersLength;
            buffer.append(characters.charAt((int) index));
        }
        return false;
    }

    private String doGet(String key) {
        String response = null;
        try {

            // Create the right URL and Http connection
                              VectorClock vc,
                              Map<String, Object> options) {
        VectorClock successfulPutVC = null;
        int expectedResponseCode = 201;
        try {
            // Create the right URL and Http connection
            HttpURLConnection conn = null;
            String base64Key = new String(Base64.encodeBase64(key.getBytes()));
            URL url = new URL(this.coordinatorURL + "/" + STORE_NAME + "/" + base64Key);
            conn = (HttpURLConnection) url.openConnection();

            // Set the right headers
            conn.setRequestMethod("POST");
            conn.setDoOutput(true);
            conn.setDoInput(true);
            conn.setRequestProperty("Content-Type", "binary");
            conn.setRequestProperty("Content-Length", "" + payload.length());
            conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_TIMEOUT_MS, "1000");
            conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_ORIGIN_TIME_MS,
                                    Long.toString(System.currentTimeMillis()));

            // options
            if(options != null) {
                if(options.get("timeout") != null && options.get("timeout") instanceof String) {
                    conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_TIMEOUT_MS,
                                            (String) options.get("timeout"));
                sb.append(line);
            }
            rd.close();
            conn.disconnect();

            response = sb.toString();

        } catch(Exception e) {
            e.printStackTrace();
            fail("Error in sending the REST request");
        }

        return response;
    }

    @Test
    public void testRESTReadAfterWrite() {
        String key = "Which_Imperial_IPA_do_I_want_to_drink";
        String payload = "Pliny the Younger";

        // 1. Do a put
        doPut(key, payload);

        // 2. Do a get on the same key
        String response = doGet(key);
        if(response == null) {
            fail("key does not exist after a put. ");
        }
        System.out.println("Received value: " + response);
        if(!response.equals(payload)) {
            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);
        }
    }

        try {

            // Create the right URL and Http connection
        String payload = "Duchesse De Bourgogne";

        // 1. Do a put
        doPut(key, payload);

        // 2. Do a get on the same key
        String response = doGet(key);
        if(response == null) {
            fail("key does not exist after a put. ");
        }
        System.out.println("Received value: " + response);
        if(!response.equals(payload)) {
            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);
        }

        // 3. Do a delete
                }
                if(options.get("responseCode") != null
                   && options.get("responseCode") instanceof Integer) {
                    expectedResponseCode = (Integer) options.get("responseCode");
                }
            }

            // Check for the right response code
            if(conn.getResponseCode() != expectedResponseCode) {
        }
    }

}
            conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_ORIGIN_TIME_MS,
                                    Long.toString(System.currentTimeMillis()));

            // options
            if(options != null) {
                if(options.get("timeout") != null && options.get("timeout") instanceof String) {
                    conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_TIMEOUT_MS,
                                            (String) options.get("timeout"));
                }
                if(options.get("responseCode") != null
                   && options.get("responseCode") instanceof Integer) {
                    expectedResponseCode = (Integer) options.get("responseCode");
                }
            }

            // Check for the right response code
            if(conn.getResponseCode() != expectedResponseCode) {
                System.err.println("Illegal response during GET : " + conn.getResponseMessage());
                fail("Incorrect response received for a HTTP GET request :"
                     + conn.getResponseCode());
            }

            if(conn.getResponseCode() == 404 || conn.getResponseCode() == 408) {
                return null;
            }

            // Buffer the result into a string
            ByteArrayDataSource ds = new ByteArrayDataSource(conn.getInputStream(),
                                                             "multipart/mixed");
            MimeMultipart mp = new MimeMultipart(ds);
            assertEquals("The number of body parts expected is not 1", 1, mp.getCount());

            MimeBodyPart part = (MimeBodyPart) mp.getBodyPart(0);
            VectorClock vc = RestUtils.deserializeVectorClock(part.getHeader(RestMessageHeaders.X_VOLD_VECTOR_CLOCK)[0]);
            int contentLength = Integer.parseInt(part.getHeader(RestMessageHeaders.CONTENT_LENGTH)[0]);
            byte[] bodyPartBytes = new byte[contentLength];

            part.getInputStream().read(bodyPartBytes);
            response = new String(bodyPartBytes);

            responseObj = new TestVersionedValue(response, vc);

        } catch(Exception e) {
            e.printStackTrace();
            fail("Error in sending the REST request");
        } finally {
            if(conn != null) {
                conn.disconnect();
            }
        }
        return responseObj;
    }

    @Test
    public void testReadAfterWrite() {
        String key = "Which_Imperial_IPA_do_I_want_to_drink";
        String payload = "Pliny the Younger";

        // 1. Do a put
        doPut(key, payload, null);

        // 2. Do a get on the same key
        TestVersionedValue response = doGet(key);
        if(response == null) {
            fail("key does not exist after a put. ");
        }

        System.out.println("Received value: " + response.getValue());
        if(!response.getValue().equals(payload)) {
            fail("Received value is incorrect ! Expected : " + payload + " but got : "
                 + response.getValue());
        }
    }

    @Test
    public void testDelete() {
        String key = "Which_sour_beer_do_I_want_to_drink";
        String payload = "Duchesse De Bourgogne";
        Map<String, Object> options = new HashMap<String, Object>();

        // 1. Do a put
        doPut(key, payload, null);

        // 2. Do a get on the same key
        TestVersionedValue response = doGet(key);
        if(response == null) {
            fail("key does not exist after a put. ");
        }
        System.out.println("Received value: " + response.getValue());
        if(!response.getValue().equals(payload)) {
            fail("Received value is incorrect ! Expected : " + payload + " but got : "
                 + response.getValue());
        }

        // 3. Do a delete
        boolean isDeleted = doDelete(key);
        if(!isDeleted) {
            fail("Could not delete the key. Error !");
        }

        // 4. Do a get on the same key : this should fail
        options.put("responseCode", 404);
        response = doGet(key, options);
        if(response != null) {
            fail("key still exists after deletion. ");
        }
    }

    @Test
    public void testVersionedPut() {
        String key = "Which_Porter_do_I_want_to_drink";
        String payload = "Founders Porter";
        String newPayload = "Samuel Smith Taddy Porter";

        // 1. Do a put
        doPut(key, payload, null);

        // 2. Do a get on the same key
        TestVersionedValue response = doGet(key, null);
        if(response == null) {
            fail("key does not exist after a put. ");
        }
        System.out.println("Received value: " + response.getValue());

        // 3. Do a versioned put based on the version received previously
        doPut(key, newPayload, response.getVc());

        // 4. Do a get again on the same key
        TestVersionedValue newResponse = doGet(key);
        if(newResponse == null) {
            fail("key does not exist after the versioned put. ");
        }
        assertEquals("Returned response does not have a higer version",
                     Occurred.AFTER,
                     newResponse.getVc().compare(response.getVc()));
        assertEquals("Returned response does not have a higer version",
                     Occurred.BEFORE,
                     response.getVc().compare(newResponse.getVc()));

        System.out.println("Received value after the Versioned put: " + newResponse.getValue());
        if(!newResponse.getValue().equals(newPayload)) {
            fail("Received value is incorrect ! Expected : " + newPayload + " but got : "
                 + newResponse.getValue());
        }
    }

    @Test
    public void testLargeValueSizeVersionedPut() {
        String key = "amigo";
        String payload = generateRandomString(new CoordinatorConfig().getHttpMessageDecoderMaxChunkSize() * 10,
                                              ValueType.ALPHA);
        String newPayload = generateRandomString(new CoordinatorConfig().getHttpMessageDecoderMaxChunkSize() * 10,
                                                 ValueType.ALPHA);

        // 1. Do a put
        doPut(key, payload, null);

        // 2. Do a get on the same key
        TestVersionedValue response = doGet(key, null);
        if(response == null) {
            fail("key does not exist after a put. ");
        }
        System.out.println("Received value: " + response.getValue());

        // 3. Do a versioned put based on the version received previously
        doPut(key, newPayload, response.getVc());

        // 4. Do a get again on the same key
        TestVersionedValue newResponse = doGet(key);
        if(newResponse == null) {
            fail("key does not exist after the versioned put. ");
        }
        assertEquals("Returned response does not have a higer version",
                     Occurred.AFTER,
                     newResponse.getVc().compare(response.getVc()));
        assertEquals("Returned response does not have a higer version",
                     Occurred.BEFORE,
                     response.getVc().compare(newResponse.getVc()));

        System.out.println("Received value after the Versioned put: " + newResponse.getValue());
        if(!newResponse.getValue().equals(newPayload)) {
            fail("Received value is incorrect ! Expected : " + newPayload + " but got : "
                 + newResponse.getValue());
        }
    }

    @Test
    public void testWriteWithTimeout() {
        String key = "Which_Imperial_IPA_do_I_want_to_drink";
        String payload = "Pliny the Younger";
        Map<String, Object> options = new HashMap<String, Object>();

        // 1. Do a put (timeout)
        options.put("timeout", "1");
        options.put("responseCode", 408);
        doPut(key, payload, null, options);

        // 2. Do a get on the same key
        options.clear();
        options.put("responseCode", 404);
        TestVersionedValue response = doGet(key, options);
        if(response != null) {
            fail("key should not exist after a put. ");
        }

        // 3. Do a put
        doPut(key, payload, null);

        // 4. Do a get on the same key with timeout
        options.clear();
        options.put("timeout", "1");
        options.put("responseCode", 408);
        response = doGet(key, options);

        // 5. Do a get on the same key
        response = doGet(key);
        if(response == null) {
            fail("key does not exist after a put. ");
        }
    }
}

<MSG> Added the ability to do Versioned puts. Added another test case for the same

<DFF> @@ -33,15 +33,14 @@ import org.junit.Before;
 import org.junit.Test;
 
 import voldemort.ServerTestUtils;
-import voldemort.cluster.Cluster;
 import voldemort.server.VoldemortServer;
 import voldemort.store.socket.SocketStoreFactory;
 import voldemort.store.socket.clientrequest.ClientRequestExecutorPool;
+import voldemort.versioning.VectorClock;
 
 public class CoordinatorRestAPITest {
 
     private VoldemortServer[] servers;
-    private Cluster cluster;
     public static String socketUrl = "";
     private static final String STORE_NAME = "test";
     private static final String STORES_XML = "test/common/voldemort/config/single-store.xml";
@@ -53,6 +52,33 @@ public class CoordinatorRestAPITest {
     private CoordinatorService coordinator = null;
     private final String coordinatorURL = "http://localhost:8080";
 
+    private class TestVersionedValue {
+
+        private String value;
+        private VectorClock vc;
+
+        public TestVersionedValue(String val, VectorClock vc) {
+            this.setValue(val);
+            this.setVc(vc);
+        }
+
+        public String getValue() {
+            return value;
+        }
+
+        public void setValue(String value) {
+            this.value = value;
+        }
+
+        public VectorClock getVc() {
+            return vc;
+        }
+
+        public void setVc(VectorClock vc) {
+            this.vc = vc;
+        }
+    }
+
     @Before
     public void setUp() throws Exception {
         int numServers = 1;
@@ -62,14 +88,14 @@ public class CoordinatorRestAPITest {
         props.setProperty("storage.configs",
                           "voldemort.store.bdb.BdbStorageConfiguration,voldemort.store.slow.SlowStorageConfiguration");
 
-        cluster = ServerTestUtils.startVoldemortCluster(numServers,
-                                                        servers,
-                                                        partitionMap,
-                                                        socketStoreFactory,
-                                                        true, // useNio
-                                                        null,
-                                                        STORES_XML,
-                                                        props);
+        ServerTestUtils.startVoldemortCluster(numServers,
+                                              servers,
+                                              partitionMap,
+                                              socketStoreFactory,
+                                              true, // useNio
+                                              null,
+                                              STORES_XML,
+                                              props);
 
         CoordinatorConfig config = new CoordinatorConfig();
         List<String> bootstrapUrls = new ArrayList<String>();
@@ -98,7 +124,8 @@ public class CoordinatorRestAPITest {
         }
     }
 
-    private void doPut(String key, String payload) {
+    private VectorClock doPut(String key, String payload, VectorClock vc) {
+        VectorClock successfulPutVC = null;
         try {
             // Create the right URL and Http connection
             HttpURLConnection conn = null;
@@ -114,6 +141,11 @@ public class CoordinatorRestAPITest {
             conn.setRequestProperty("Content-Length", "" + payload.length());
             conn.setRequestProperty(VoldemortHttpRequestHandler.X_VOLD_REQUEST_TIMEOUT_MS, "1000");
 
+            if(vc != null) {
+                String eTag = CoordinatorUtils.getSerializedVectorClock(vc);
+                conn.setRequestProperty("ETag", eTag);
+            }
+
             // Write the payload
             OutputStream out = conn.getOutputStream();
             out.write(payload.getBytes());
@@ -125,10 +157,13 @@ public class CoordinatorRestAPITest {
                 fail("Incorrect response received for a HTTP put request :"
                      + conn.getResponseCode());
             }
+
         } catch(Exception e) {
             e.printStackTrace();
             fail("Error in sending the REST request");
         }
+
+        return successfulPutVC;
     }
 
     private boolean doDelete(String key) {
@@ -162,8 +197,9 @@ public class CoordinatorRestAPITest {
         return false;
     }
 
-    private String doGet(String key) {
+    private TestVersionedValue doGet(String key) {
         String response = null;
+        TestVersionedValue responseObj = null;
         try {
 
             // Create the right URL and Http connection
@@ -196,34 +232,39 @@ public class CoordinatorRestAPITest {
                 sb.append(line);
             }
             rd.close();
+
             conn.disconnect();
 
             response = sb.toString();
+            VectorClock vc = CoordinatorUtils.deserializeVectorClock(conn.getHeaderField("ETag"));
+            responseObj = new TestVersionedValue(response, vc);
 
         } catch(Exception e) {
             e.printStackTrace();
             fail("Error in sending the REST request");
         }
 
-        return response;
+        return responseObj;
     }
 
     @Test
-    public void testRESTReadAfterWrite() {
+    public void testReadAfterWrite() {
         String key = "Which_Imperial_IPA_do_I_want_to_drink";
         String payload = "Pliny the Younger";
 
         // 1. Do a put
-        doPut(key, payload);
+        doPut(key, payload, null);
 
         // 2. Do a get on the same key
-        String response = doGet(key);
+        TestVersionedValue response = doGet(key);
         if(response == null) {
             fail("key does not exist after a put. ");
         }
-        System.out.println("Received value: " + response);
-        if(!response.equals(payload)) {
-            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);
+
+        System.out.println("Received value: " + response.getValue());
+        if(!response.getValue().equals(payload)) {
+            fail("Received value is incorrect ! Expected : " + payload + " but got : "
+                 + response.getValue());
         }
     }
 
@@ -233,16 +274,17 @@ public class CoordinatorRestAPITest {
         String payload = "Duchesse De Bourgogne";
 
         // 1. Do a put
-        doPut(key, payload);
+        doPut(key, payload, null);
 
         // 2. Do a get on the same key
-        String response = doGet(key);
+        TestVersionedValue response = doGet(key);
         if(response == null) {
             fail("key does not exist after a put. ");
         }
-        System.out.println("Received value: " + response);
-        if(!response.equals(payload)) {
-            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);
+        System.out.println("Received value: " + response.getValue());
+        if(!response.getValue().equals(payload)) {
+            fail("Received value is incorrect ! Expected : " + payload + " but got : "
+                 + response.getValue());
         }
 
         // 3. Do a delete
@@ -258,4 +300,35 @@ public class CoordinatorRestAPITest {
         }
     }
 
+    @Test
+    public void testVersionedPut() {
+        String key = "Which_Porter_do_I_want_to_drink";
+        String payload = "Founders Porter";
+        String newPayload = "Samuel Smith Taddy Porter";
+
+        // 1. Do a put
+        doPut(key, payload, null);
+
+        // 2. Do a get on the same key
+        TestVersionedValue response = doGet(key);
+        if(response == null) {
+            fail("key does not exist after a put. ");
+        }
+        System.out.println("Received value: " + response.getValue());
+
+        // 3. Do a versioned put based on the version received previously
+        doPut(key, newPayload, response.getVc());
+
+        // 4. Do a get again on the same key
+        TestVersionedValue newResponse = doGet(key);
+        if(newResponse == null) {
+            fail("key does not exist after the versioned put. ");
+        }
+
+        System.out.println("Received value after the Versioned put: " + newResponse.getValue());
+        if(!newResponse.getValue().equals(newPayload)) {
+            fail("Received value is incorrect ! Expected : " + newPayload + " but got : "
+                 + newResponse.getValue());
+        }
+    }
 }

{'commit': 'diff --git a/contrib/restclient/src/java/voldemort/restclient/R2Store.java b/contrib/restclient/src/java/voldemort/restclient/R2Store.java\nindex 4b2abc26f..6cc084bf0 100644\n--- a/contrib/restclient/src/java/voldemort/restclient/R2Store.java\n+++ b/contrib/restclient/src/java/voldemort/restclient/R2Store.java\n@@ -41,6 +41,7 @@ import org.codehaus.jackson.map.JsonMappingException;\n import org.codehaus.jackson.map.ObjectMapper;\n \n import voldemort.VoldemortException;\n+import voldemort.coordinator.CoordinatorUtils;\n import voldemort.coordinator.VectorClockWrapper;\n import voldemort.store.AbstractStore;\n import voldemort.utils.ByteArray;\n@@ -70,10 +71,12 @@ public class R2Store extends AbstractStore<ByteArray, byte[], byte[]> {\n     private static final String POST = "POST";\n     private static final String DELETE = "DELETE";\n     private static final String ETAG = "ETag";\n+    public static final String X_VOLD_VECTOR_CLOCK = "X-VOLD-Vector-Clock";\n     public static final String X_VOLD_REQUEST_TIMEOUT_MS = "X-VOLD-Request-Timeout-ms";\n     public static final String X_VOLD_INCONSISTENCY_RESOLVER = "X-VOLD-Inconsistency-Resolver";\n     public static final String CUSTOM_RESOLVING_STRATEGY = "custom";\n     public static final String DEFAULT_RESOLVING_STRATEGY = "timestamp";\n+\n     private static final String LAST_MODIFIED = "Last-Modified";\n     private static final String MULTIPART_CONTENT_TYPE = "multipart/binary";\n     private final Logger logger = Logger.getLogger(R2Store.class);\n@@ -202,6 +205,9 @@ public class R2Store extends AbstractStore<ByteArray, byte[], byte[]> {\n             RestRequestBuilder rb = new RestRequestBuilder(new URI(this.baseURL + "/" + getName()\n                                                                    + "/" + base64Key));\n \n+            // Serialize the Vector clock\n+            String serializedVC = CoordinatorUtils.getSerializedVectorClock((VectorClock) value.getVersion());\n+\n             // Create a HTTP POST request\n             // TODO: Create a proper request based on client config\n             rb.setMethod(POST);\n@@ -209,13 +215,14 @@ public class R2Store extends AbstractStore<ByteArray, byte[], byte[]> {\n             rb.setHeader("Content-Type", "application/json");\n             rb.setHeader("Content-Length", "" + payload.length);\n             rb.setHeader(X_VOLD_REQUEST_TIMEOUT_MS, "1000");\n-            rb.setHeader(X_VOLD_INCONSISTENCY_RESOLVER, "custom");\n+            rb.setHeader(X_VOLD_VECTOR_CLOCK, serializedVC);\n \n             RestRequest request = rb.build();\n             Future<RestResponse> f = client.restRequest(request);\n \n             // This will block\n             RestResponse response = f.get();\n+            String eTag = response.getHeader(ETAG);\n             final ByteString entity = response.getEntity();\n             if(entity == null) {\n                 logger.error("Empty response !");\ndiff --git a/contrib/restclient/src/java/voldemort/restclient/SampleRESTClient.java b/contrib/restclient/src/java/voldemort/restclient/SampleRESTClient.java\nindex 595351117..4eb512469 100644\n--- a/contrib/restclient/src/java/voldemort/restclient/SampleRESTClient.java\n+++ b/contrib/restclient/src/java/voldemort/restclient/SampleRESTClient.java\n@@ -19,6 +19,8 @@ package voldemort.restclient;\n import java.util.ArrayList;\n import java.util.List;\n \n+import voldemort.versioning.Versioned;\n+\n public class SampleRESTClient {\n \n     public static void main(String[] args) {\n@@ -31,8 +33,18 @@ public class SampleRESTClient {\n         clientStore.put("a", "Howdy!!!!");\n         clientStore.put("b", "Partner!!!!");\n \n-        // Do a sample operation:\n-        System.out.println("Received response : " + clientStore.get("a"));\n+        // Do a sample get operation:\n+        Versioned<String> versionedValue = clientStore.get("a");\n+        System.out.println("Received response : " + versionedValue);\n+\n+        // Do a versioned put operation:\n+        versionedValue.setObject("New Value !!!");\n+        clientStore.put("a", versionedValue);\n+\n+        // Do a get again on the last versioned put operation:\n+        versionedValue = clientStore.get("a");\n+        System.out.println("Received response on the versioned put: " + versionedValue);\n+\n         List<String> keyList = new ArrayList<String>();\n         keyList.add("a");\n         keyList.add("b");\ndiff --git a/src/java/voldemort/coordinator/CoordinatorUtils.java b/src/java/voldemort/coordinator/CoordinatorUtils.java\nnew file mode 100644\nindex 000000000..d8003ca65\n--- /dev/null\n+++ b/src/java/voldemort/coordinator/CoordinatorUtils.java\n@@ -0,0 +1,46 @@\n+package voldemort.coordinator;\n+\n+import org.codehaus.jackson.map.ObjectMapper;\n+\n+import voldemort.versioning.VectorClock;\n+\n+public class CoordinatorUtils {\n+\n+    /**\n+     * Function to serialize the given Vector clock into a string. If something\n+     * goes wrong, it returns an empty string.\n+     * \n+     * @param vc The Vector clock to serialize\n+     * @return The string (JSON) version of the specified Vector clock\n+     */\n+    public static String getSerializedVectorClock(VectorClock vc) {\n+        VectorClockWrapper vcWrapper = new VectorClockWrapper(vc);\n+        ObjectMapper mapper = new ObjectMapper();\n+        String serializedVC = "";\n+        try {\n+            serializedVC = mapper.writeValueAsString(vcWrapper);\n+        } catch(Exception e) {\n+            e.printStackTrace();\n+        }\n+        return serializedVC;\n+    }\n+\n+    public static VectorClock deserializeVectorClock(String serializedVC) {\n+        VectorClock vc = null;\n+\n+        if(serializedVC == null) {\n+            return null;\n+        }\n+\n+        ObjectMapper mapper = new ObjectMapper();\n+\n+        try {\n+            VectorClockWrapper vcWrapper = mapper.readValue(serializedVC, VectorClockWrapper.class);\n+            vc = new VectorClock(vcWrapper.getVersions(), vcWrapper.getTimestamp());\n+        } catch(Exception e) {\n+            e.printStackTrace();\n+        }\n+\n+        return vc;\n+    }\n+}\ndiff --git a/src/java/voldemort/coordinator/HttpGetRequestExecutor.java b/src/java/voldemort/coordinator/HttpGetRequestExecutor.java\nindex c87490977..02a1de9a5 100644\n--- a/src/java/voldemort/coordinator/HttpGetRequestExecutor.java\n+++ b/src/java/voldemort/coordinator/HttpGetRequestExecutor.java\n@@ -27,12 +27,7 @@ import static org.jboss.netty.handler.codec.http.HttpResponseStatus.OK;\n import static org.jboss.netty.handler.codec.http.HttpResponseStatus.REQUEST_TIMEOUT;\n import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1;\n \n-import java.io.IOException;\n-\n import org.apache.log4j.Logger;\n-import org.codehaus.jackson.JsonGenerationException;\n-import org.codehaus.jackson.map.JsonMappingException;\n-import org.codehaus.jackson.map.ObjectMapper;\n import org.jboss.netty.buffer.ChannelBuffer;\n import org.jboss.netty.buffer.ChannelBuffers;\n import org.jboss.netty.channel.MessageEvent;\n@@ -110,18 +105,7 @@ public class HttpGetRequestExecutor implements Runnable {\n         this.responseContent.writeBytes(value);\n \n         VectorClock vc = (VectorClock) responseVersioned.getVersion();\n-        VectorClockWrapper vcWrapper = new VectorClockWrapper(vc);\n-        ObjectMapper mapper = new ObjectMapper();\n-        String eTag = "";\n-        try {\n-            eTag = mapper.writeValueAsString(vcWrapper);\n-        } catch(JsonGenerationException e) {\n-            e.printStackTrace();\n-        } catch(JsonMappingException e) {\n-            e.printStackTrace();\n-        } catch(IOException e) {\n-            e.printStackTrace();\n-        }\n+        String eTag = CoordinatorUtils.getSerializedVectorClock(vc);\n \n         if(logger.isDebugEnabled()) {\n             logger.debug("ETAG : " + eTag);\ndiff --git a/src/java/voldemort/coordinator/HttpPutRequestExecutor.java b/src/java/voldemort/coordinator/HttpPutRequestExecutor.java\nindex e3d69ff8d..e0ad0f98d 100644\n--- a/src/java/voldemort/coordinator/HttpPutRequestExecutor.java\n+++ b/src/java/voldemort/coordinator/HttpPutRequestExecutor.java\n@@ -17,6 +17,7 @@\n package voldemort.coordinator;\n \n import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH;\n+import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.ETAG;\n import static org.jboss.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;\n import static org.jboss.netty.handler.codec.http.HttpResponseStatus.CREATED;\n import static org.jboss.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;\n@@ -36,6 +37,7 @@ import voldemort.store.stats.StoreStats;\n import voldemort.store.stats.Tracked;\n import voldemort.utils.ByteArray;\n import voldemort.versioning.ObsoleteVersionException;\n+import voldemort.versioning.VectorClock;\n \n /**\n  * A Runnable class that uses the specified Fat client to perform a Voldemort\n@@ -88,11 +90,18 @@ public class HttpPutRequestExecutor implements Runnable {\n         this.coordinatorPerfStats = coordinatorPerfStats;\n     }\n \n-    public void writeResponse() {\n+    public void writeResponse(VectorClock successfulPutVC) {\n         // 1. Create the Response object\n         HttpResponse response = new DefaultHttpResponse(HTTP_1_1, CREATED);\n \n+        String eTag = CoordinatorUtils.getSerializedVectorClock(successfulPutVC);\n+\n+        if(logger.isDebugEnabled()) {\n+            logger.debug("ETAG : " + eTag);\n+        }\n+\n         // 2. Set the right headers\n+        response.setHeader(ETAG, eTag);\n         response.setHeader(CONTENT_LENGTH, 0);\n \n         // TODO: return the Version back to the client\n@@ -111,11 +120,16 @@ public class HttpPutRequestExecutor implements Runnable {\n     public void run() {\n \n         try {\n-            this.storeClient.putWithCustomTimeout(putRequestObject);\n+            VectorClock successfulPutVC = null;\n+            if(putRequestObject.getValue() != null) {\n+                successfulPutVC = (VectorClock) this.storeClient.putVersionedWithCustomTimeout(putRequestObject);\n+            } else {\n+                successfulPutVC = (VectorClock) this.storeClient.putWithCustomTimeout(putRequestObject);\n+            }\n             if(logger.isDebugEnabled()) {\n                 logger.debug("PUT successful !");\n             }\n-            writeResponse();\n+            writeResponse(successfulPutVC);\n \n         } catch(IllegalArgumentException illegalArgsException) {\n             String errorDescription = "PUT Failed !!! Illegal Arguments : "\ndiff --git a/src/java/voldemort/coordinator/NoopHttpRequestHandler.java b/src/java/voldemort/coordinator/NoopHttpRequestHandler.java\nindex 0c0ceb56a..79a07b4a3 100644\n--- a/src/java/voldemort/coordinator/NoopHttpRequestHandler.java\n+++ b/src/java/voldemort/coordinator/NoopHttpRequestHandler.java\n@@ -22,6 +22,7 @@ import org.jboss.netty.handler.codec.http.HttpMethod;\n import org.jboss.netty.handler.codec.http.HttpRequest;\n \n import voldemort.common.VoldemortOpCode;\n+import voldemort.versioning.VectorClock;\n import voldemort.versioning.Versioned;\n \n /**\n@@ -50,7 +51,7 @@ public class NoopHttpRequestHandler extends VoldemortHttpRequestHandler {\n                 break;\n             case VoldemortOpCode.PUT_OP_CODE:\n                 HttpPutRequestExecutor putRequestExecutor = new HttpPutRequestExecutor(e);\n-                putRequestExecutor.writeResponse();\n+                putRequestExecutor.writeResponse(new VectorClock());\n                 break;\n             default:\n                 System.err.println("Illegal operation.");\ndiff --git a/src/java/voldemort/coordinator/VoldemortHttpRequestHandler.java b/src/java/voldemort/coordinator/VoldemortHttpRequestHandler.java\nindex fe10519a8..a6b9553a4 100644\n--- a/src/java/voldemort/coordinator/VoldemortHttpRequestHandler.java\n+++ b/src/java/voldemort/coordinator/VoldemortHttpRequestHandler.java\n@@ -42,9 +42,11 @@ import voldemort.store.CompositeDeleteVoldemortRequest;\n import voldemort.store.CompositeGetAllVoldemortRequest;\n import voldemort.store.CompositeGetVoldemortRequest;\n import voldemort.store.CompositePutVoldemortRequest;\n+import voldemort.store.CompositeVersionedPutVoldemortRequest;\n import voldemort.store.CompositeVoldemortRequest;\n import voldemort.utils.ByteArray;\n import voldemort.versioning.VectorClock;\n+import voldemort.versioning.Versioned;\n \n /**\n  * A class to handle the HTTP request and execute the same on behalf of the thin\n@@ -153,15 +155,23 @@ public class VoldemortHttpRequestHandler extends SimpleChannelUpstreamHandler {\n                     return null;\n                 }\n                 byte[] putValue = readValue(content);\n-                requestWrapper = new CompositePutVoldemortRequest<ByteArray, byte[]>(putKey,\n-                                                                                     putValue,\n-                                                                                     operationTimeoutInMs);\n+                VectorClock putOpVectorClock = getVectorClock(this.request.getHeader(X_VOLD_VECTOR_CLOCK));\n+                if(putOpVectorClock != null && putOpVectorClock.getEntries().size() > 0) {\n+                    requestWrapper = new CompositeVersionedPutVoldemortRequest<ByteArray, byte[]>(putKey,\n+                                                                                                  new Versioned<byte[]>(putValue,\n+                                                                                                                        putOpVectorClock),\n+                                                                                                  operationTimeoutInMs);\n+                } else {\n+                    requestWrapper = new CompositePutVoldemortRequest<ByteArray, byte[]>(putKey,\n+                                                                                         putValue,\n+                                                                                         operationTimeoutInMs);\n+                }\n \n                 break;\n             case VoldemortOpCode.DELETE_OP_CODE:\n-                VectorClock vc = getVectorClock(this.request.getHeader(X_VOLD_VECTOR_CLOCK));\n+                VectorClock deleteOpVectorClock = getVectorClock(this.request.getHeader(X_VOLD_VECTOR_CLOCK));\n                 requestWrapper = new CompositeDeleteVoldemortRequest<ByteArray, byte[]>(keyList.get(0),\n-                                                                                        vc,\n+                                                                                        deleteOpVectorClock,\n                                                                                         operationTimeoutInMs);\n \n                 break;\n@@ -262,6 +272,11 @@ public class VoldemortHttpRequestHandler extends SimpleChannelUpstreamHandler {\n      */\n     private VectorClock getVectorClock(String vectorClockHeader) {\n         VectorClock vc = null;\n+\n+        if(vectorClockHeader == null) {\n+            return null;\n+        }\n+\n         ObjectMapper mapper = new ObjectMapper();\n         if(logger.isDebugEnabled()) {\n             logger.debug("Received vector clock : " + vectorClockHeader);\ndiff --git a/test/unit/voldemort/coordinator/CoordinatorRestAPITest.java b/test/unit/voldemort/coordinator/CoordinatorRestAPITest.java\nindex 496e6379b..17ce89b76 100644\n--- a/test/unit/voldemort/coordinator/CoordinatorRestAPITest.java\n+++ b/test/unit/voldemort/coordinator/CoordinatorRestAPITest.java\n@@ -33,15 +33,14 @@ import org.junit.Before;\n import org.junit.Test;\n \n import voldemort.ServerTestUtils;\n-import voldemort.cluster.Cluster;\n import voldemort.server.VoldemortServer;\n import voldemort.store.socket.SocketStoreFactory;\n import voldemort.store.socket.clientrequest.ClientRequestExecutorPool;\n+import voldemort.versioning.VectorClock;\n \n public class CoordinatorRestAPITest {\n \n     private VoldemortServer[] servers;\n-    private Cluster cluster;\n     public static String socketUrl = "";\n     private static final String STORE_NAME = "test";\n     private static final String STORES_XML = "test/common/voldemort/config/single-store.xml";\n@@ -53,6 +52,33 @@ public class CoordinatorRestAPITest {\n     private CoordinatorService coordinator = null;\n     private final String coordinatorURL = "http://localhost:8080";\n \n+    private class TestVersionedValue {\n+\n+        private String value;\n+        private VectorClock vc;\n+\n+        public TestVersionedValue(String val, VectorClock vc) {\n+            this.setValue(val);\n+            this.setVc(vc);\n+        }\n+\n+        public String getValue() {\n+            return value;\n+        }\n+\n+        public void setValue(String value) {\n+            this.value = value;\n+        }\n+\n+        public VectorClock getVc() {\n+            return vc;\n+        }\n+\n+        public void setVc(VectorClock vc) {\n+            this.vc = vc;\n+        }\n+    }\n+\n     @Before\n     public void setUp() throws Exception {\n         int numServers = 1;\n@@ -62,14 +88,14 @@ public class CoordinatorRestAPITest {\n         props.setProperty("storage.configs",\n                           "voldemort.store.bdb.BdbStorageConfiguration,voldemort.store.slow.SlowStorageConfiguration");\n \n-        cluster = ServerTestUtils.startVoldemortCluster(numServers,\n-                                                        servers,\n-                                                        partitionMap,\n-                                                        socketStoreFactory,\n-                                                        true, // useNio\n-                                                        null,\n-                                                        STORES_XML,\n-                                                        props);\n+        ServerTestUtils.startVoldemortCluster(numServers,\n+                                              servers,\n+                                              partitionMap,\n+                                              socketStoreFactory,\n+                                              true, // useNio\n+                                              null,\n+                                              STORES_XML,\n+                                              props);\n \n         CoordinatorConfig config = new CoordinatorConfig();\n         List<String> bootstrapUrls = new ArrayList<String>();\n@@ -98,7 +124,8 @@ public class CoordinatorRestAPITest {\n         }\n     }\n \n-    private void doPut(String key, String payload) {\n+    private VectorClock doPut(String key, String payload, VectorClock vc) {\n+        VectorClock successfulPutVC = null;\n         try {\n             // Create the right URL and Http connection\n             HttpURLConnection conn = null;\n@@ -114,6 +141,11 @@ public class CoordinatorRestAPITest {\n             conn.setRequestProperty("Content-Length", "" + payload.length());\n             conn.setRequestProperty(VoldemortHttpRequestHandler.X_VOLD_REQUEST_TIMEOUT_MS, "1000");\n \n+            if(vc != null) {\n+                String eTag = CoordinatorUtils.getSerializedVectorClock(vc);\n+                conn.setRequestProperty("ETag", eTag);\n+            }\n+\n             // Write the payload\n             OutputStream out = conn.getOutputStream();\n             out.write(payload.getBytes());\n@@ -125,10 +157,13 @@ public class CoordinatorRestAPITest {\n                 fail("Incorrect response received for a HTTP put request :"\n                      + conn.getResponseCode());\n             }\n+\n         } catch(Exception e) {\n             e.printStackTrace();\n             fail("Error in sending the REST request");\n         }\n+\n+        return successfulPutVC;\n     }\n \n     private boolean doDelete(String key) {\n@@ -162,8 +197,9 @@ public class CoordinatorRestAPITest {\n         return false;\n     }\n \n-    private String doGet(String key) {\n+    private TestVersionedValue doGet(String key) {\n         String response = null;\n+        TestVersionedValue responseObj = null;\n         try {\n \n             // Create the right URL and Http connection\n@@ -196,34 +232,39 @@ public class CoordinatorRestAPITest {\n                 sb.append(line);\n             }\n             rd.close();\n+\n             conn.disconnect();\n \n             response = sb.toString();\n+            VectorClock vc = CoordinatorUtils.deserializeVectorClock(conn.getHeaderField("ETag"));\n+            responseObj = new TestVersionedValue(response, vc);\n \n         } catch(Exception e) {\n             e.printStackTrace();\n             fail("Error in sending the REST request");\n         }\n \n-        return response;\n+        return responseObj;\n     }\n \n     @Test\n-    public void testRESTReadAfterWrite() {\n+    public void testReadAfterWrite() {\n         String key = "Which_Imperial_IPA_do_I_want_to_drink";\n         String payload = "Pliny the Younger";\n \n         // 1. Do a put\n-        doPut(key, payload);\n+        doPut(key, payload, null);\n \n         // 2. Do a get on the same key\n-        String response = doGet(key);\n+        TestVersionedValue response = doGet(key);\n         if(response == null) {\n             fail("key does not exist after a put. ");\n         }\n-        System.out.println("Received value: " + response);\n-        if(!response.equals(payload)) {\n-            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);\n+\n+        System.out.println("Received value: " + response.getValue());\n+        if(!response.getValue().equals(payload)) {\n+            fail("Received value is incorrect ! Expected : " + payload + " but got : "\n+                 + response.getValue());\n         }\n     }\n \n@@ -233,16 +274,17 @@ public class CoordinatorRestAPITest {\n         String payload = "Duchesse De Bourgogne";\n \n         // 1. Do a put\n-        doPut(key, payload);\n+        doPut(key, payload, null);\n \n         // 2. Do a get on the same key\n-        String response = doGet(key);\n+        TestVersionedValue response = doGet(key);\n         if(response == null) {\n             fail("key does not exist after a put. ");\n         }\n-        System.out.println("Received value: " + response);\n-        if(!response.equals(payload)) {\n-            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);\n+        System.out.println("Received value: " + response.getValue());\n+        if(!response.getValue().equals(payload)) {\n+            fail("Received value is incorrect ! Expected : " + payload + " but got : "\n+                 + response.getValue());\n         }\n \n         // 3. Do a delete\n@@ -258,4 +300,35 @@ public class CoordinatorRestAPITest {\n         }\n     }\n \n+    @Test\n+    public void testVersionedPut() {\n+        String key = "Which_Porter_do_I_want_to_drink";\n+        String payload = "Founders Porter";\n+        String newPayload = "Samuel Smith Taddy Porter";\n+\n+        // 1. Do a put\n+        doPut(key, payload, null);\n+\n+        // 2. Do a get on the same key\n+        TestVersionedValue response = doGet(key);\n+        if(response == null) {\n+            fail("key does not exist after a put. ");\n+        }\n+        System.out.println("Received value: " + response.getValue());\n+\n+        // 3. Do a versioned put based on the version received previously\n+        doPut(key, newPayload, response.getVc());\n+\n+        // 4. Do a get again on the same key\n+        TestVersionedValue newResponse = doGet(key);\n+        if(newResponse == null) {\n+            fail("key does not exist after the versioned put. ");\n+        }\n+\n+        System.out.println("Received value after the Versioned put: " + newResponse.getValue());\n+        if(!newResponse.getValue().equals(newPayload)) {\n+            fail("Received value is incorrect ! Expected : " + newPayload + " but got : "\n+                 + newResponse.getValue());\n+        }\n+    }\n }\n', 'commit_subject': 'Added the ability to do Versioned puts. Added another test case for the same', 'repo_name': 'voldemort/voldemort', 'committer': {'name': 'Chinmay Soman', 'email': '742c8f6c91b89b837bfc0b99918e7763cd7a78d2@linkedin.com', 'time_sec': '1367001871', 'tz_offset': '-420', 'date': {'seconds': '1367001871'}}, 'commit_tree': '743c3fcb92c7c839dfffe1e9f646fbb203324418', 'commit_message': 'Added the ability to do Versioned puts. Added another test case for the same\n', 'language_name': 'Python', 'license': 'apache-2.0', 'hunks': '@@ -33,15 +33,14 @@ import org.junit.Before;\n import org.junit.Test;\n \n import voldemort.ServerTestUtils;\n-import voldemort.cluster.Cluster;\n import voldemort.server.VoldemortServer;\n import voldemort.store.socket.SocketStoreFactory;\n import voldemort.store.socket.clientrequest.ClientRequestExecutorPool;\n+import voldemort.versioning.VectorClock;\n \n public class CoordinatorRestAPITest {\n \n     private VoldemortServer[] servers;\n-    private Cluster cluster;\n     public static String socketUrl = "";\n     private static final String STORE_NAME = "test";\n     private static final String STORES_XML = "test/common/voldemort/config/single-store.xml";\n@@ -53,6 +52,33 @@ public class CoordinatorRestAPITest {\n     private CoordinatorService coordinator = null;\n     private final String coordinatorURL = "http://localhost:8080";\n \n+    private class TestVersionedValue {\n+\n+        private String value;\n+        private VectorClock vc;\n+\n+        public TestVersionedValue(String val, VectorClock vc) {\n+            this.setValue(val);\n+            this.setVc(vc);\n+        }\n+\n+        public String getValue() {\n+            return value;\n+        }\n+\n+        public void setValue(String value) {\n+            this.value = value;\n+        }\n+\n+        public VectorClock getVc() {\n+            return vc;\n+        }\n+\n+        public void setVc(VectorClock vc) {\n+            this.vc = vc;\n+        }\n+    }\n+\n     @Before\n     public void setUp() throws Exception {\n         int numServers = 1;\n@@ -62,14 +88,14 @@ public class CoordinatorRestAPITest {\n         props.setProperty("storage.configs",\n                           "voldemort.store.bdb.BdbStorageConfiguration,voldemort.store.slow.SlowStorageConfiguration");\n \n-        cluster = ServerTestUtils.startVoldemortCluster(numServers,\n-                                                        servers,\n-                                                        partitionMap,\n-                                                        socketStoreFactory,\n-                                                        true, // useNio\n-                                                        null,\n-                                                        STORES_XML,\n-                                                        props);\n+        ServerTestUtils.startVoldemortCluster(numServers,\n+                                              servers,\n+                                              partitionMap,\n+                                              socketStoreFactory,\n+                                              true, // useNio\n+                                              null,\n+                                              STORES_XML,\n+                                              props);\n \n         CoordinatorConfig config = new CoordinatorConfig();\n         List<String> bootstrapUrls = new ArrayList<String>();\n@@ -98,7 +124,8 @@ public class CoordinatorRestAPITest {\n         }\n     }\n \n-    private void doPut(String key, String payload) {\n+    private VectorClock doPut(String key, String payload, VectorClock vc) {\n+        VectorClock successfulPutVC = null;\n         try {\n             // Create the right URL and Http connection\n             HttpURLConnection conn = null;\n@@ -114,6 +141,11 @@ public class CoordinatorRestAPITest {\n             conn.setRequestProperty("Content-Length", "" + payload.length());\n             conn.setRequestProperty(VoldemortHttpRequestHandler.X_VOLD_REQUEST_TIMEOUT_MS, "1000");\n \n+            if(vc != null) {\n+                String eTag = CoordinatorUtils.getSerializedVectorClock(vc);\n+                conn.setRequestProperty("ETag", eTag);\n+            }\n+\n             // Write the payload\n             OutputStream out = conn.getOutputStream();\n             out.write(payload.getBytes());\n@@ -125,10 +157,13 @@ public class CoordinatorRestAPITest {\n                 fail("Incorrect response received for a HTTP put request :"\n                      + conn.getResponseCode());\n             }\n+\n         } catch(Exception e) {\n             e.printStackTrace();\n             fail("Error in sending the REST request");\n         }\n+\n+        return successfulPutVC;\n     }\n \n     private boolean doDelete(String key) {\n@@ -162,8 +197,9 @@ public class CoordinatorRestAPITest {\n         return false;\n     }\n \n-    private String doGet(String key) {\n+    private TestVersionedValue doGet(String key) {\n         String response = null;\n+        TestVersionedValue responseObj = null;\n         try {\n \n             // Create the right URL and Http connection\n@@ -196,34 +232,39 @@ public class CoordinatorRestAPITest {\n                 sb.append(line);\n             }\n             rd.close();\n+\n             conn.disconnect();\n \n             response = sb.toString();\n+            VectorClock vc = CoordinatorUtils.deserializeVectorClock(conn.getHeaderField("ETag"));\n+            responseObj = new TestVersionedValue(response, vc);\n \n         } catch(Exception e) {\n             e.printStackTrace();\n             fail("Error in sending the REST request");\n         }\n \n-        return response;\n+        return responseObj;\n     }\n \n     @Test\n-    public void testRESTReadAfterWrite() {\n+    public void testReadAfterWrite() {\n         String key = "Which_Imperial_IPA_do_I_want_to_drink";\n         String payload = "Pliny the Younger";\n \n         // 1. Do a put\n-        doPut(key, payload);\n+        doPut(key, payload, null);\n \n         // 2. Do a get on the same key\n-        String response = doGet(key);\n+        TestVersionedValue response = doGet(key);\n         if(response == null) {\n             fail("key does not exist after a put. ");\n         }\n-        System.out.println("Received value: " + response);\n-        if(!response.equals(payload)) {\n-            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);\n+\n+        System.out.println("Received value: " + response.getValue());\n+        if(!response.getValue().equals(payload)) {\n+            fail("Received value is incorrect ! Expected : " + payload + " but got : "\n+                 + response.getValue());\n         }\n     }\n \n@@ -233,16 +274,17 @@ public class CoordinatorRestAPITest {\n         String payload = "Duchesse De Bourgogne";\n \n         // 1. Do a put\n-        doPut(key, payload);\n+        doPut(key, payload, null);\n \n         // 2. Do a get on the same key\n-        String response = doGet(key);\n+        TestVersionedValue response = doGet(key);\n         if(response == null) {\n             fail("key does not exist after a put. ");\n         }\n-        System.out.println("Received value: " + response);\n-        if(!response.equals(payload)) {\n-            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);\n+        System.out.println("Received value: " + response.getValue());\n+        if(!response.getValue().equals(payload)) {\n+            fail("Received value is incorrect ! Expected : " + payload + " but got : "\n+                 + response.getValue());\n         }\n \n         // 3. Do a delete\n@@ -258,4 +300,35 @@ public class CoordinatorRestAPITest {\n         }\n     }\n \n+    @Test\n+    public void testVersionedPut() {\n+        String key = "Which_Porter_do_I_want_to_drink";\n+        String payload = "Founders Porter";\n+        String newPayload = "Samuel Smith Taddy Porter";\n+\n+        // 1. Do a put\n+        doPut(key, payload, null);\n+\n+        // 2. Do a get on the same key\n+        TestVersionedValue response = doGet(key);\n+        if(response == null) {\n+            fail("key does not exist after a put. ");\n+        }\n+        System.out.println("Received value: " + response.getValue());\n+\n+        // 3. Do a versioned put based on the version received previously\n+        doPut(key, newPayload, response.getVc());\n+\n+        // 4. Do a get again on the same key\n+        TestVersionedValue newResponse = doGet(key);\n+        if(newResponse == null) {\n+            fail("key does not exist after the versioned put. ");\n+        }\n+\n+        System.out.println("Received value after the Versioned put: " + newResponse.getValue());\n+        if(!newResponse.getValue().equals(newPayload)) {\n+            fail("Received value is incorrect ! Expected : " + newPayload + " but got : "\n+                 + newResponse.getValue());\n+        }\n+    }\n }\n', 'addition_count': 97, 'deletion_count': 24, 'src_file': 'a/test/unit/voldemort/coordinator/CoordinatorRestAPITest.java', 'tgt_file': 'b/test/unit/voldemort/coordinator/CoordinatorRestAPITest.java', 'file_extension': '.java', 'before_file': '/*\n * Copyright 2013 LinkedIn, Inc\n * \n * Licensed under the Apache License, Version 2.0 (the "License"); you may not\n * use this file except in compliance with the License. You may obtain a copy of\n * the License at\n * \n * http://www.apache.org/licenses/LICENSE-2.0\n * \n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\n * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n * License for the specific language governing permissions and limitations under\n * the License.\n */\n\npackage voldemort.coordinator;\n\nimport static org.junit.Assert.assertEquals;\nimport static org.junit.Assert.fail;\n\nimport java.io.File;\nimport java.io.OutputStream;\nimport java.net.HttpURLConnection;\nimport java.net.URL;\nimport java.util.ArrayList;\nimport java.util.HashMap;\nimport java.util.List;\nimport java.util.Map;\nimport java.util.Properties;\n\nimport javax.mail.internet.MimeBodyPart;\nimport org.junit.Test;\n\nimport voldemort.ServerTestUtils;\nimport voldemort.cluster.Cluster;\nimport voldemort.server.VoldemortServer;\nimport voldemort.store.socket.SocketStoreFactory;\nimport voldemort.store.socket.clientrequest.ClientRequestExecutorPool;\n\npublic class CoordinatorRestAPITest {\n\n    private VoldemortServer[] servers;\n    private Cluster cluster;\n    public static String socketUrl = "";\n    private static final String STORE_NAME = "test";\n    private static final String STORES_XML = "test/common/voldemort/config/single-store.xml";\nimport voldemort.rest.coordinator.config.CoordinatorConfig;\nimport voldemort.rest.coordinator.config.FileBasedStoreClientConfigService;\nimport voldemort.rest.coordinator.config.StoreClientConfigService;\nimport voldemort.server.VoldemortServer;\nimport voldemort.store.socket.SocketStoreFactory;\n    private CoordinatorService coordinator = null;\n    private final String coordinatorURL = "http://localhost:8080";\n\n    @Before\n    public void setUp() throws Exception {\n        int numServers = 1;\n        public void setValue(String value) {\n            this.value = value;\n        }\n        props.setProperty("storage.configs",\n                          "voldemort.store.bdb.BdbStorageConfiguration,voldemort.store.slow.SlowStorageConfiguration");\n\n        cluster = ServerTestUtils.startVoldemortCluster(numServers,\n                                                        servers,\n                                                        partitionMap,\n                                                        socketStoreFactory,\n                                                        true, // useNio\n                                                        null,\n                                                        STORES_XML,\n                                                        props);\n\n        CoordinatorConfig config = new CoordinatorConfig();\n        List<String> bootstrapUrls = new ArrayList<String>();\n        int partitionMap[][] = { { 0, 2, 4, 6, 1, 3, 5, 7 } };\n        Properties props = new Properties();\n        props.setProperty("storage.configs",\n                          "voldemort.store.bdb.BdbStorageConfiguration,voldemort.store.slow.SlowStorageConfiguration");\n        props.setProperty("testing.slow.queueing.get.ms", "4");\n        props.setProperty("testing.slow.queueing.getall.ms", "4");\n        props.setProperty("testing.slow.queueing.put.ms", "4");\n        props.setProperty("testing.slow.queueing.delete.ms", "4");\n\n        ServerTestUtils.startVoldemortCluster(numServers,\n                                              servers,\n                                              partitionMap,\n                                              socketStoreFactory,\n                                              true, // useNio\n                                              null,\n                                              STORES_XML,\n                                              props);\n\n        // create a copy of the config file in a temp directory and work on that\n        File src = new File(FAT_CLIENT_CONFIG_FILE_PATH_ORIGINAL);\n        COPY_OF_FAT_CLIENT_CONFIG_FILE = new File(TestUtils.createTempDir(),\n                                                  "fat-client-config" + System.currentTimeMillis()\n        }\n    }\n\n    private void doPut(String key, String payload) {\n        try {\n            // Create the right URL and Http connection\n            HttpURLConnection conn = null;\n        System.out.println("\\n\\n************************ Starting the Coordinator *************************");\n\n        config.setBootstrapURLs(bootstrapUrls);\n        config.setFatClientConfigPath(COPY_OF_FAT_CLIENT_CONFIG_FILE.getAbsolutePath());\n        StoreClientConfigService storeClientConfigs = null;\n        switch(config.getFatClientConfigSource()) {\n            case FILE:\n                storeClientConfigs = new FileBasedStoreClientConfigService(config);\n                break;\n            conn.setRequestProperty("Content-Length", "" + payload.length());\n            conn.setRequestProperty(VoldemortHttpRequestHandler.X_VOLD_REQUEST_TIMEOUT_MS, "1000");\n\n            // Write the payload\n            OutputStream out = conn.getOutputStream();\n            out.write(payload.getBytes());\n    @After\n    public void tearDown() throws Exception {\n        if(this.socketStoreFactory != null) {\n            this.socketStoreFactory.close();\n        }\n                fail("Incorrect response received for a HTTP put request :"\n                     + conn.getResponseCode());\n            }\n        } catch(Exception e) {\n            e.printStackTrace();\n            fail("Error in sending the REST request");\n        }\n    }\n\n    private boolean doDelete(String key) {\n\n    public static String generateRandomString(int length, ValueType type) {\n\n        StringBuffer buffer = new StringBuffer();\n        String characters = "";\n\n        switch(type) {\n\n            case ALPHA:\n                characters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";\n                break;\n\n            case ALPHANUMERIC:\n                characters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890";\n                break;\n\n            case NUMERIC:\n                characters = "1234567890";\n                break;\n        }\n\n        int charactersLength = characters.length();\n\n        for(int i = 0; i < length; i++) {\n            double index = Math.random() * charactersLength;\n            buffer.append(characters.charAt((int) index));\n        }\n        return false;\n    }\n\n    private String doGet(String key) {\n        String response = null;\n        try {\n\n            // Create the right URL and Http connection\n                              VectorClock vc,\n                              Map<String, Object> options) {\n        VectorClock successfulPutVC = null;\n        int expectedResponseCode = 201;\n        try {\n            // Create the right URL and Http connection\n            HttpURLConnection conn = null;\n            String base64Key = new String(Base64.encodeBase64(key.getBytes()));\n            URL url = new URL(this.coordinatorURL + "/" + STORE_NAME + "/" + base64Key);\n            conn = (HttpURLConnection) url.openConnection();\n\n            // Set the right headers\n            conn.setRequestMethod("POST");\n            conn.setDoOutput(true);\n            conn.setDoInput(true);\n            conn.setRequestProperty("Content-Type", "binary");\n            conn.setRequestProperty("Content-Length", "" + payload.length());\n            conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_TIMEOUT_MS, "1000");\n            conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_ORIGIN_TIME_MS,\n                                    Long.toString(System.currentTimeMillis()));\n\n            // options\n            if(options != null) {\n                if(options.get("timeout") != null && options.get("timeout") instanceof String) {\n                    conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_TIMEOUT_MS,\n                                            (String) options.get("timeout"));\n                sb.append(line);\n            }\n            rd.close();\n            conn.disconnect();\n\n            response = sb.toString();\n\n        } catch(Exception e) {\n            e.printStackTrace();\n            fail("Error in sending the REST request");\n        }\n\n        return response;\n    }\n\n    @Test\n    public void testRESTReadAfterWrite() {\n        String key = "Which_Imperial_IPA_do_I_want_to_drink";\n        String payload = "Pliny the Younger";\n\n        // 1. Do a put\n        doPut(key, payload);\n\n        // 2. Do a get on the same key\n        String response = doGet(key);\n        if(response == null) {\n            fail("key does not exist after a put. ");\n        }\n        System.out.println("Received value: " + response);\n        if(!response.equals(payload)) {\n            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);\n        }\n    }\n\n        try {\n\n            // Create the right URL and Http connection\n        String payload = "Duchesse De Bourgogne";\n\n        // 1. Do a put\n        doPut(key, payload);\n\n        // 2. Do a get on the same key\n        String response = doGet(key);\n        if(response == null) {\n            fail("key does not exist after a put. ");\n        }\n        System.out.println("Received value: " + response);\n        if(!response.equals(payload)) {\n            fail("Received value is incorrect ! Expected : " + payload + " but got : " + response);\n        }\n\n        // 3. Do a delete\n                }\n                if(options.get("responseCode") != null\n                   && options.get("responseCode") instanceof Integer) {\n                    expectedResponseCode = (Integer) options.get("responseCode");\n                }\n            }\n\n            // Check for the right response code\n            if(conn.getResponseCode() != expectedResponseCode) {\n        }\n    }\n\n}\n            conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_ORIGIN_TIME_MS,\n                                    Long.toString(System.currentTimeMillis()));\n\n            // options\n            if(options != null) {\n                if(options.get("timeout") != null && options.get("timeout") instanceof String) {\n                    conn.setRequestProperty(RestMessageHeaders.X_VOLD_REQUEST_TIMEOUT_MS,\n                                            (String) options.get("timeout"));\n                }\n                if(options.get("responseCode") != null\n                   && options.get("responseCode") instanceof Integer) {\n                    expectedResponseCode = (Integer) options.get("responseCode");\n                }\n            }\n\n            // Check for the right response code\n            if(conn.getResponseCode() != expectedResponseCode) {\n                System.err.println("Illegal response during GET : " + conn.getResponseMessage());\n                fail("Incorrect response received for a HTTP GET request :"\n                     + conn.getResponseCode());\n            }\n\n            if(conn.getResponseCode() == 404 || conn.getResponseCode() == 408) {\n                return null;\n            }\n\n            // Buffer the result into a string\n            ByteArrayDataSource ds = new ByteArrayDataSource(conn.getInputStream(),\n                                                             "multipart/mixed");\n            MimeMultipart mp = new MimeMultipart(ds);\n            assertEquals("The number of body parts expected is not 1", 1, mp.getCount());\n\n            MimeBodyPart part = (MimeBodyPart) mp.getBodyPart(0);\n            VectorClock vc = RestUtils.deserializeVectorClock(part.getHeader(RestMessageHeaders.X_VOLD_VECTOR_CLOCK)[0]);\n            int contentLength = Integer.parseInt(part.getHeader(RestMessageHeaders.CONTENT_LENGTH)[0]);\n            byte[] bodyPartBytes = new byte[contentLength];\n\n            part.getInputStream().read(bodyPartBytes);\n            response = new String(bodyPartBytes);\n\n            responseObj = new TestVersionedValue(response, vc);\n\n        } catch(Exception e) {\n            e.printStackTrace();\n            fail("Error in sending the REST request");\n        } finally {\n            if(conn != null) {\n                conn.disconnect();\n            }\n        }\n        return responseObj;\n    }\n\n    @Test\n    public void testReadAfterWrite() {\n        String key = "Which_Imperial_IPA_do_I_want_to_drink";\n        String payload = "Pliny the Younger";\n\n        // 1. Do a put\n        doPut(key, payload, null);\n\n        // 2. Do a get on the same key\n        TestVersionedValue response = doGet(key);\n        if(response == null) {\n            fail("key does not exist after a put. ");\n        }\n\n        System.out.println("Received value: " + response.getValue());\n        if(!response.getValue().equals(payload)) {\n            fail("Received value is incorrect ! Expected : " + payload + " but got : "\n                 + response.getValue());\n        }\n    }\n\n    @Test\n    public void testDelete() {\n        String key = "Which_sour_beer_do_I_want_to_drink";\n        String payload = "Duchesse De Bourgogne";\n        Map<String, Object> options = new HashMap<String, Object>();\n\n        // 1. Do a put\n        doPut(key, payload, null);\n\n        // 2. Do a get on the same key\n        TestVersionedValue response = doGet(key);\n        if(response == null) {\n            fail("key does not exist after a put. ");\n        }\n        System.out.println("Received value: " + response.getValue());\n        if(!response.getValue().equals(payload)) {\n            fail("Received value is incorrect ! Expected : " + payload + " but got : "\n                 + response.getValue());\n        }\n\n        // 3. Do a delete\n        boolean isDeleted = doDelete(key);\n        if(!isDeleted) {\n            fail("Could not delete the key. Error !");\n        }\n\n        // 4. Do a get on the same key : this should fail\n        options.put("responseCode", 404);\n        response = doGet(key, options);\n        if(response != null) {\n            fail("key still exists after deletion. ");\n        }\n    }\n\n    @Test\n    public void testVersionedPut() {\n        String key = "Which_Porter_do_I_want_to_drink";\n        String payload = "Founders Porter";\n        String newPayload = "Samuel Smith Taddy Porter";\n\n        // 1. Do a put\n        doPut(key, payload, null);\n\n        // 2. Do a get on the same key\n        TestVersionedValue response = doGet(key, null);\n        if(response == null) {\n            fail("key does not exist after a put. ");\n        }\n        System.out.println("Received value: " + response.getValue());\n\n        // 3. Do a versioned put based on the version received previously\n        doPut(key, newPayload, response.getVc());\n\n        // 4. Do a get again on the same key\n        TestVersionedValue newResponse = doGet(key);\n        if(newResponse == null) {\n            fail("key does not exist after the versioned put. ");\n        }\n        assertEquals("Returned response does not have a higer version",\n                     Occurred.AFTER,\n                     newResponse.getVc().compare(response.getVc()));\n        assertEquals("Returned response does not have a higer version",\n                     Occurred.BEFORE,\n                     response.getVc().compare(newResponse.getVc()));\n\n        System.out.println("Received value after the Versioned put: " + newResponse.getValue());\n        if(!newResponse.getValue().equals(newPayload)) {\n            fail("Received value is incorrect ! Expected : " + newPayload + " but got : "\n                 + newResponse.getValue());\n        }\n    }\n\n    @Test\n    public void testLargeValueSizeVersionedPut() {\n        String key = "amigo";\n        String payload = generateRandomString(new CoordinatorConfig().getHttpMessageDecoderMaxChunkSize() * 10,\n                                              ValueType.ALPHA);\n        String newPayload = generateRandomString(new CoordinatorConfig().getHttpMessageDecoderMaxChunkSize() * 10,\n                                                 ValueType.ALPHA);\n\n        // 1. Do a put\n        doPut(key, payload, null);\n\n        // 2. Do a get on the same key\n        TestVersionedValue response = doGet(key, null);\n        if(response == null) {\n            fail("key does not exist after a put. ");\n        }\n        System.out.println("Received value: " + response.getValue());\n\n        // 3. Do a versioned put based on the version received previously\n        doPut(key, newPayload, response.getVc());\n\n        // 4. Do a get again on the same key\n        TestVersionedValue newResponse = doGet(key);\n        if(newResponse == null) {\n            fail("key does not exist after the versioned put. ");\n        }\n        assertEquals("Returned response does not have a higer version",\n                     Occurred.AFTER,\n                     newResponse.getVc().compare(response.getVc()));\n        assertEquals("Returned response does not have a higer version",\n                     Occurred.BEFORE,\n                     response.getVc().compare(newResponse.getVc()));\n\n        System.out.println("Received value after the Versioned put: " + newResponse.getValue());\n        if(!newResponse.getValue().equals(newPayload)) {\n            fail("Received value is incorrect ! Expected : " + newPayload + " but got : "\n                 + newResponse.getValue());\n        }\n    }\n\n    @Test\n    public void testWriteWithTimeout() {\n        String key = "Which_Imperial_IPA_do_I_want_to_drink";\n        String payload = "Pliny the Younger";\n        Map<String, Object> options = new HashMap<String, Object>();\n\n        // 1. Do a put (timeout)\n        options.put("timeout", "1");\n        options.put("responseCode", 408);\n        doPut(key, payload, null, options);\n\n        // 2. Do a get on the same key\n        options.clear();\n        options.put("responseCode", 404);\n        TestVersionedValue response = doGet(key, options);\n        if(response != null) {\n            fail("key should not exist after a put. ");\n        }\n\n        // 3. Do a put\n        doPut(key, payload, null);\n\n        // 4. Do a get on the same key with timeout\n        options.clear();\n        options.put("timeout", "1");\n        options.put("responseCode", 408);\n        response = doGet(key, options);\n\n        // 5. Do a get on the same key\n        response = doGet(key);\n        if(response == null) {\n            fail("key does not exist after a put. ");\n        }\n    }\n}\n', 'message': 'Added the ability to do Versioned puts. Added another test case for the same\n', 'author_name': 'Chinmay Soman', 'lang': 'java'}