Addressing code review comments.

src/java/voldemort/client/protocol/admin/AdminClient.java

@@ -2097,7 +2097,31 @@ public void updateEntries(int nodeId,
                                   String storeName,
                                   Iterator<Pair<ByteArray, Versioned<byte[]>>> entryIterator,
                                   VoldemortFilter filter) {
-            updateEntries(nodeId, storeName, entryIterator, filter, false);
+            streamingUpdateEntries(nodeId, storeName, entryIterator, filter, false);
+        }
+
+        /**
+         * Update a stream of key/value entries at the given node in the same
+         * way as
+         * {@link StreamingOperations#updateEntries(int, String, Iterator, VoldemortFilter)}
+         * 
+         * The only difference being the resolving on the server will happen
+         * based on timestamp and not the vector clock.
+         * 
+         * @param nodeId Id of the remote node (where we wish to update the
+         *        entries)
+         * @param storeName Store name for the entries
+         * @param entryIterator Iterator of key-value pairs for the entries
+         * @param filter Custom filter implementation to filter out entries
+         *        which should not be updated.
+         * 
+         * @throws VoldemortException
+         */
+        public void updateEntriesTimeBased(int nodeId,
+                                           String storeName,
+                                           Iterator<Pair<ByteArray, Versioned<byte[]>>> entryIterator,
+                                           VoldemortFilter filter) {
+            streamingUpdateEntries(nodeId, storeName, entryIterator, filter, true);
         }
 
         /**
@@ -2121,15 +2145,16 @@ public void updateEntries(int nodeId,
          * @param entryIterator Iterator of key-value pairs for the entries
          * @param filter Custom filter implementation to filter out entries
          *        which should not be updated.
-         * @param overWriteIfLatest overwrite the existing value if the supplied
-         *        version has greater timestamp
+         * @param overWriteIfLatestTs if true overwrite the existing value if
+         *        the supplied version has greater timestamp; else use vector
+         *        clocks
          * @throws VoldemortException
          */
-        public void updateEntries(int nodeId,
-                                  String storeName,
-                                  Iterator<Pair<ByteArray, Versioned<byte[]>>> entryIterator,
-                                  VoldemortFilter filter,
-                                  boolean overWriteIfLatest) {
+        private void streamingUpdateEntries(int nodeId,
+                                            String storeName,
+                                            Iterator<Pair<ByteArray, Versioned<byte[]>>> entryIterator,
+                                            VoldemortFilter filter,
+                                            boolean overWriteIfLatestTs) {
             Node node = AdminClient.this.getAdminClientCluster().getNodeById(nodeId);
             SocketDestination destination = new SocketDestination(node.getHost(),
                                                                   node.getAdminPort(),
@@ -2151,11 +2176,11 @@ public void updateEntries(int nodeId,
                                                                                               .build();
                         VAdminProto.UpdatePartitionEntriesRequest.Builder updateRequest = null;
 
-                        if(overWriteIfLatest) {
+                        if(overWriteIfLatestTs) {
                             updateRequest = VAdminProto.UpdatePartitionEntriesRequest.newBuilder()
                                                                                      .setStore(storeName)
                                                                                      .setPartitionEntry(partitionEntry)
-                                                                                     .setOverwriteIfLatest(overWriteIfLatest);
+                                                                                     .setOverwriteIfLatestTs(overWriteIfLatestTs);
                         } else {
                             updateRequest = VAdminProto.UpdatePartitionEntriesRequest.newBuilder()
                                                                                      .setStore(storeName)

src/java/voldemort/server/protocol/admin/AdminServiceRequestHandler.java

@@ -558,7 +558,7 @@ public StreamRequestHandler handleUpdatePartitionEntries(VAdminProto.UpdateParti
         StorageEngine<ByteArray, byte[], byte[]> storageEngine = AdminServiceRequestHandler.getStorageEngine(storeRepository,
                                                                                                              request.getStore());
 
-        if(request.hasOverwriteIfLatest() && request.getOverwriteIfLatest()) {
+        if(request.hasOverwriteIfLatestTs() && request.getOverwriteIfLatestTs()) {
             // Resolve based on timestamp if specified.
             return new TimeBasedUpdatePartitionEntriesStreamRequestHandler(request,
                                                                            errorCodeMapper,

src/java/voldemort/server/protocol/admin/FetchPartitionFileStreamRequestHandler.java

@@ -82,10 +82,8 @@ protected FetchPartitionFileStreamRequestHandler(VAdminProto.FetchPartitionFiles
                                                      StoreRepository storeRepository) {
         this.request = request;
         // TODO (Sid) : Confirm if keeping metadatastore as a class property is
-        // the best
-        // way to do this. metadataStore is used later in the
-        // handleNextPartition() to create
-        // object plan.
+        // the best way to do this. metadataStore is used later in the
+        // handleNextPartition() to create object plan.
         this.metadataStore = metadataStore;
 
         StoreDefinition storeDef = metadataStore.getStoreDef(request.getStoreName());

src/java/voldemort/server/protocol/admin/TimeBasedUpdatePartitionEntriesStreamRequestHandler.java

@@ -9,7 +9,6 @@
 import voldemort.server.VoldemortConfig;
 import voldemort.server.storage.KeyLockHandle;
 import voldemort.store.ErrorCodeMapper;
-import voldemort.store.PersistenceFailureException;
 import voldemort.store.StorageEngine;
 import voldemort.utils.ByteArray;
 import voldemort.utils.NetworkClassLoader;
@@ -76,12 +75,12 @@ protected void processEntry(ByteArray key, Versioned<byte[]> value) throws IOExc
                 // greater than what we are trying to write in
                 storageEngine.releaseLock(handle);
             }
-        } catch(PersistenceFailureException pfe) {
+        } catch(Exception e) {
             if(handle != null && !handle.isClosed()) {
                 storageEngine.releaseLock(handle);
             }
-            logger.error("Error in time based update entries", pfe);
-            throw new IOException(pfe);
+            logger.error("Error in time based update entries", e);
+            throw new IOException(e);
         }
     }
 

src/java/voldemort/server/protocol/admin/UpdatePartitionEntriesStreamRequestHandler.java

@@ -26,10 +26,7 @@
 import voldemort.utils.ByteUtils;
 import voldemort.utils.EventThrottler;
 import voldemort.utils.NetworkClassLoader;
-<<<<<<< HEAD
-=======
 import voldemort.utils.Time;
->>>>>>> Timestamp based resolving for UpdateEntries
 import voldemort.utils.Utils;
 import voldemort.versioning.ObsoleteVersionException;
 import voldemort.versioning.Versioned;
@@ -38,8 +35,14 @@
  * UpdatePartitionEntriesStreamRequestHandler implements the streaming logic for
  * updating partition entries.
  * 
- * This is the base class, which simply reads a versioned entry from network and
- * issues a storage engine put.
+ * This is the base class, which abstracts network IO to get an entry off the
+ * wire, and provides a hook
+ * {@link UpdatePartitionEntriesStreamRequestHandler#processEntry(ByteArray, Versioned)}
+ * to implement custom logic (if needed) to manage how the entry will be written
+ * to storage.
+ * 
+ * The default implementation of processEntry(..) simply issues a storage engine
+ * put.
  * 
  */
 

src/java/voldemort/store/StorageEngine.java

@@ -138,10 +138,14 @@ public interface StorageEngine<K, V, T> extends Store<K, V, T> {
     /**
      * Returns the list of versions stored for the key, at the same time locking
      * the key for any writes until
-     * {@link StorageEngine#putAndUnlock(Object, KeyLockHandle)} is called with
-     * the same lock handle. The idea here is to facilitate custom atomic
+     * {@link StorageEngine#putAndUnlock(Object, KeyLockHandle)} or
+     * {@link StorageEngine#releaseLock(KeyLockHandle)} is called with the same
+     * lock handle. The idea here is to facilitate custom atomic
      * Read-Modify-Write logic outside the storage engine
      * 
+     * NOTE : An invocation of getAndLock should be followed by EXACTLY ONE call
+     * to either putAndLock or releaseLock, for resources to be freed properly
+     * 
      * @param key
      * @return
      */
@@ -160,7 +164,8 @@ public interface StorageEngine<K, V, T> extends Store<K, V, T> {
 
     /**
      * Release any lock held by a prior
-     * {@link AbstractStorageEngine#getAndLock(Object)} call
+     * {@link AbstractStorageEngine#getAndLock(Object)} call. Helpful for
+     * exception handling during a read-modify-cycle
      * 
      * @param handle
      */

src/java/voldemort/store/bdb/BdbStorageEngine.java

@@ -690,6 +690,9 @@ public KeyLockHandle<byte[]> getAndLock(ByteArray key) {
             handle = new KeyLockHandle<byte[]>(vals, transaction);
         } catch(DatabaseException e) {
             this.bdbEnvironmentStats.reportException(e);
+            // Unless we return out properly from this method, we need to ensure
+            // the transaction handle is closed on exception..
+            attemptAbort(transaction);
             logger.error("Error in getAndLock for store " + this.getName(), e);
             throw new PersistenceFailureException(e);
         } finally {

src/proto/voldemort-admin.proto

@@ -43,7 +43,7 @@ message UpdatePartitionEntriesRequest {
   required string store = 1;
   required PartitionEntry partition_entry = 2;
   optional VoldemortFilter filter = 3;
-  optional bool overwriteIfLatest = 4;
+  optional bool overwriteIfLatestTs = 4;
 }
 
 message UpdatePartitionEntriesResponse {

test/unit/voldemort/client/AdminServiceBasicTest.java

@@ -1587,7 +1587,7 @@ protected Pair<ByteArray, Versioned<byte[]>> computeNext() {
             }
         };
 
-        getAdminClient().streamingOps.updateEntries(0, storeName, iterator, null, true);
+        getAdminClient().streamingOps.updateEntriesTimeBased(0, storeName, iterator, null);
 
         // check updated values
         for(int i = 0; i < keys.size(); i++) {